note記事データをpandasで分析する｜伸びる型を数字で見つける

"""
記事メタデータとPV統計を結合し、分析用のDataFrameを作る。
"""
import os
import re

import pandas as pd

ARTICLES_CSV = "note_articles.csv"
PV_CSV = "pv_history.csv"


def load_articles(path: str = ARTICLES_CSV) -> pd.DataFrame:
    df = pd.read_csv(path, encoding="utf-8-sig")

    df["published_at"] = pd.to_datetime(
        df["published_at"], errors="coerce", utc=True
    ).dt.tz_convert("Asia/Tokyo")

    return df


def load_latest_pv(path: str = PV_CSV) -> pd.DataFrame:
    """PV履歴から、最新時点の数字だけを取り出す"""
    if not os.path.exists(path):
        return pd.DataFrame(columns=["key", "pv"])

    df = pd.read_csv(path, encoding="utf-8-sig")
    df["date"] = pd.to_datetime(df["date"])

    latest = df[df["date"] == df["date"].max()]
    return latest[["key", "pv"]]


def add_features(df: pd.DataFrame) -> pd.DataFrame:
    """分析に使う派生列を追加する"""
    df = df.copy()

    # 日時から曜日・時間帯
    df["weekday"] = df["published_at"].dt.dayofweek     # 0=月曜
    df["weekday_name"] = df["published_at"].dt.day_name()
    df["hour"] = df["published_at"].dt.hour
    df["year_month"] = df["published_at"].dt.strftime("%Y-%m")

    # 公開からの経過日数
    now = pd.Timestamp.now(tz="Asia/Tokyo")
    df["days_since"] = (now - df["published_at"]).dt.days

    # タイトルの特徴
    title = df["title"].astype(str)
    df["title_len"] = title.str.len()
    df["has_number"] = title.str.contains(r"\d").astype(int)
    df["has_question"] = title.str.contains(r"[？?]").astype(int)
    df["has_bracket"] = title.str.contains(r"[【\[]").astype(int)
    df["has_howto"] = title.str.contains(
        "方法|やり方|手順|コツ|使い方"
    ).astype(int)
    df["has_number_list"] = title.str.contains(
        r"\d+\s*(選|個|つ|の理由|のコツ|ステップ)"
    ).astype(int)

    # 有料かどうか
    df["is_paid"] = (df["price"].fillna(0) > 0).astype(int)

    return df


def build(articles_path: str = ARTICLES_CSV,
          pv_path: str = PV_CSV) -> pd.DataFrame:
    """分析用のDataFrameを組み立てる"""
    df = load_articles(articles_path)
    pv = load_latest_pv(pv_path)

    if not pv.empty:
        df = df.merge(pv, on="key", how="left")
        df["pv"] = df["pv"].fillna(0).astype(int)
    else:
        df["pv"] = 0
        print("PVデータがありません。スキ数のみで分析します")

    df = add_features(df)

    # 1日あたりのPV（公開からの日数で割る）
    df["pv_per_day"] = (
        df["pv"] / df["days_since"].clip(lower=1)
    ).round(2)

    return df


if __name__ == "__main__":
    df = build()

    print(f"=== 記事 {len(df)} 本 ===")
    print(f"  期間: {df['published_at'].min().date()} 〜 "
          f"{df['published_at'].max().date()}")
    print(f"  総PV: {df['pv'].sum():,}")
    print(f"  総スキ: {df['likes'].sum():,}")

    print("\n=== 列の一覧 ===")
    print(", ".join(df.columns))

"""
タイトルの特徴ごとに、反応の差を比較する。
"""
import pandas as pd

from dataset import build

FEATURES = {
    "has_number": "数字を含む",
    "has_question": "疑問形",
    "has_bracket": "括弧を含む",
    "has_howto": "方法・やり方系",
    "has_number_list": "数字＋選/個/つ",
}


def compare_feature(df: pd.DataFrame, column: str,
                    label: str, metric: str = "pv_per_day"):
    """特徴の有無で、指標の平均を比較する"""
    with_feature = df[df[column] == 1][metric]
    without = df[df[column] == 0][metric]

    if len(with_feature) < 3 or len(without) < 3:
        print(f"  {label}: サンプルが少なく比較できません")
        return

    avg_with = with_feature.mean()
    avg_without = without.mean()
    ratio = avg_with / avg_without if avg_without else 0

    mark = "↑" if ratio > 1.1 else ("↓" if ratio < 0.9 else "→")

    print(f"  {label}")
    print(f"    あり（{len(with_feature):3}本）: {avg_with:8.2f}")
    print(f"    なし（{len(without):3}本）: {avg_without:8.2f}")
    print(f"    比率: {ratio:.2f} 倍 {mark}")


def analyze_length(df: pd.DataFrame, metric: str = "pv_per_day"):
    """タイトルの長さ帯ごとの平均"""
    bins = [0, 15, 20, 25, 30, 40, 999]
    labels = ["〜15", "16-20", "21-25", "26-30", "31-40", "41〜"]

    df = df.copy()
    df["len_band"] = pd.cut(df["title_len"], bins=bins, labels=labels)

    result = df.groupby("len_band", observed=True).agg(
        count=("key", "count"),
        avg=(metric, "mean"),
    ).round(2)

    print("\n=== タイトルの長さ別 ===")
    for band, row in result.iterrows():
        if row["count"] < 2:
            continue
        bar = "|" * int(min(row["avg"] * 3, 40))
        print(f"  {band:>6}文字  {int(row['count']):3}本  "
              f"{row['avg']:7.2f}  {bar}")


def correlation(df: pd.DataFrame):
    """数値列同士の相関を見る"""
    cols = ["title_len", "likes", "pv", "pv_per_day", "days_since"]
    available = [c for c in cols if c in df.columns]

    corr = df[available].corr().round(3)

    print("\n=== 相関係数 ===")
    print(corr.to_string())
    print("\n  ※ 0.3以上で弱い関連、0.5以上でやや強い関連")
    print("  ※ 相関があっても、因果があるとは限らない")


def main():
    df = build()

    metric = "pv_per_day" if df["pv"].sum() > 0 else "likes"
    print(f"=== タイトルの特徴別の比較（指標: {metric}）===\n")

    for column, label in FEATURES.items():
        compare_feature(df, column, label, metric)
        print()

    analyze_length(df, metric)
    correlation(df)


if __name__ == "__main__":
    main()

"""
公開した曜日・時間帯ごとの反応を集計する。
"""
import pandas as pd

from dataset import build

WEEKDAY_JP = ["月", "火", "水", "木", "金", "土", "日"]

TIME_BANDS = [
    (0, 6, "深夜 0-6時"),
    (6, 9, "早朝 6-9時"),
    (9, 12, "午前 9-12時"),
    (12, 15, "昼 12-15時"),
    (15, 18, "夕方 15-18時"),
    (18, 21, "夜 18-21時"),
    (21, 24, "深夜前 21-24時"),
]


def band_of(hour: int) -> str:
    for start, end, label in TIME_BANDS:
        if start <= hour < end:
            return label
    return "不明"


def by_weekday(df: pd.DataFrame, metric: str):
    result = df.groupby("weekday").agg(
        count=("key", "count"),
        avg=(metric, "mean"),
    ).round(2)

    print("=== 曜日別 ===")
    for weekday, row in result.iterrows():
        if pd.isna(weekday):
            continue
        name = WEEKDAY_JP[int(weekday)]
        bar = "|" * int(min(row["avg"] * 3, 40))
        note = "（サンプル少）" if row["count"] < 5 else ""
        print(f"  {name}曜  {int(row['count']):3}本  "
              f"{row['avg']:7.2f}  {bar} {note}")


def by_time_band(df: pd.DataFrame, metric: str):
    df = df.copy()
    df["band"] = df["hour"].apply(
        lambda h: band_of(int(h)) if pd.notna(h) else "不明"
    )

    result = df.groupby("band").agg(
        count=("key", "count"),
        avg=(metric, "mean"),
    ).round(2)

    print("\n=== 時間帯別 ===")
    for _, _, label in TIME_BANDS:
        if label not in result.index:
            continue
        row = result.loc[label]
        bar = "|" * int(min(row["avg"] * 3, 40))
        note = "（サンプル少）" if row["count"] < 5 else ""
        print(f"  {label:>12}  {int(row['count']):3}本  "
              f"{row['avg']:7.2f}  {bar} {note}")


def by_month(df: pd.DataFrame, metric: str):
    result = df.groupby("year_month").agg(
        count=("key", "count"),
        avg=(metric, "mean"),
        total_likes=("likes", "sum"),
    ).round(2)

    print("\n=== 月別の投稿ペース ===")
    for month, row in result.tail(12).iterrows():
        bar = "|" * int(row["count"])
        print(f"  {month}  {int(row['count']):3}本  {bar}")


def main():
    df = build()
    metric = "pv_per_day" if df["pv"].sum() > 0 else "likes"

    print(f"指標: {metric}\n")

    by_weekday(df, metric)
    by_time_band(df, metric)
    by_month(df, metric)

    print("\n【解釈の注意】")
    print("  投稿時間の影響は、検索流入が中心の記事ではほぼ無い。")
    print("  差が出た場合も、その時間に良い記事を書いていただけ")
    print("  という可能性を先に疑うこと。")


if __name__ == "__main__":
    main()

"""
伸びた記事と伸びなかった記事で、タイトルの語彙を比較する。
形態素解析なしで、傾向をつかむ。
"""
import re
from collections import Counter

import pandas as pd

from dataset import build

STOP_WORDS = {
    "する", "した", "して", "こと", "もの", "ため", "よう",
    "です", "ます", "から", "まで", "note", "自分",
}


def tokenize(title: str) -> list:
    """漢字・カタカナ・英字のまとまりを拾う"""
    if not isinstance(title, str):
        return []

    words = []
    words += re.findall(r"[一-龥]{2,}", title)
    words += re.findall(r"[ァ-ヶー]{3,}", title)
    words += re.findall(r"[a-zA-Z]{3,}", title)

    return [w for w in words if w not in STOP_WORDS]


def compare_vocabulary(df: pd.DataFrame, metric: str,
                       ratio: float = 0.3):
    """上位グループと下位グループの語彙を比較する"""
    df = df.sort_values(metric, ascending=False)
    n = max(int(len(df) * ratio), 5)

    top = df.head(n)
    bottom = df.tail(n)

    top_words = Counter()
    for t in top["title"]:
        top_words.update(tokenize(t))

    bottom_words = Counter()
    for t in bottom["title"]:
        bottom_words.update(tokenize(t))

    print(f"=== 上位{n}本によく出る語 ===")
    for word, count in top_words.most_common(20):
        in_bottom = bottom_words.get(word, 0)
        mark = " ★" if count >= 2 and in_bottom == 0 else ""
        print(f"  {word:<14} 上位{count:2}回 / 下位{in_bottom:2}回{mark}")

    print(f"\n=== 下位{n}本にだけ出る語 ===")
    only_bottom = [
        (w, c) for w, c in bottom_words.most_common(30)
        if top_words.get(w, 0) == 0 and c >= 2
    ]

    for word, count in only_bottom[:15]:
        print(f"  {word:<14} {count} 回")

    print("\n  ★ = 上位にだけ出現している語")
    print("  ※ サンプルが少ないと偶然の可能性が高い")


def show_extremes(df: pd.DataFrame, metric: str, n: int = 10):
    """上位と下位の記事タイトルを並べて見る"""
    df = df.sort_values(metric, ascending=False)

    print(f"\n=== 上位{n}本 ===")
    for _, row in df.head(n).iterrows():
        print(f"  {row[metric]:8.2f}  {str(row['title'])[:48]}")

    print(f"\n=== 下位{n}本 ===")
    for _, row in df.tail(n).iterrows():
        print(f"  {row[metric]:8.2f}  {str(row['title'])[:48]}")


def main():
    df = build()
    metric = "pv_per_day" if df["pv"].sum() > 0 else "likes"

    if len(df) < 20:
        print("記事数が少ないため、傾向の判断は難しい状態です")

    compare_vocabulary(df, metric)
    show_extremes(df, metric)


if __name__ == "__main__":
    main()

"""
有料記事と無料記事を比較する。
"""
import pandas as pd

from dataset import build


def compare(df: pd.DataFrame):
    paid = df[df["is_paid"] == 1]
    free = df[df["is_paid"] == 0]

    if paid.empty:
        print("有料記事がありません")
        return

    print(f"=== 有料 {len(paid)} 本 / 無料 {len(free)} 本 ===\n")

    metrics = [
        ("pv", "累計PV"),
        ("pv_per_day", "1日あたりPV"),
        ("likes", "スキ数"),
        ("title_len", "タイトル文字数"),
    ]

    print(f"{'指標':<16}{'有料':>12}{'無料':>12}")
    print("-" * 42)

    for column, label in metrics:
        if column not in df.columns:
            continue
        p = paid[column].mean()
        f = free[column].mean() if not free.empty else 0
        print(f"{label:<16}{p:>12.2f}{f:>12.2f}")

    print("\n=== 価格帯の分布 ===")
    bins = [0, 300, 500, 1000, 2000, 5000, 10 ** 9]
    labels = ["〜300", "301-500", "501-1000",
              "1001-2000", "2001-5000", "5001〜"]

    paid = paid.copy()
    paid["band"] = pd.cut(paid["price"], bins=bins, labels=labels)

    result = paid.groupby("band", observed=True).agg(
        count=("key", "count"),
        avg_likes=("likes", "mean"),
    ).round(1)

    for band, row in result.iterrows():
        bar = "|" * int(row["count"])
        print(f"  {band:>12}円  {int(row['count']):3}本  {bar}")

    print("\n=== 有料記事の一覧 ===")
    for _, row in paid.sort_values("price").iterrows():
        print(f"  {int(row['price']):>6,}円  スキ{int(row['likes']):>4}  "
              f"{str(row['title'])[:40]}")

    print("\n【注意】")
    print("  APIからは販売数が取得できないため、")
    print("  売れているかどうかはこの分析では判断できない。")
    print("  スキ数は関心の目安にはなるが、購入とは別の指標。")


def main():
    df = build()
    compare(df)


if __name__ == "__main__":
    main()

"""
分析結果をまとめてテキストレポートに出力する。
定期的に実行して、前回との変化を見る。
"""
from datetime import datetime

import pandas as pd

from dataset import build
from title_analysis import FEATURES

OUTPUT = "analysis_report.md"


def build_report(df: pd.DataFrame) -> str:
    metric = "pv_per_day" if df["pv"].sum() > 0 else "likes"

    lines = [
        "# note 記事分析レポート",
        "",
        f"作成日時: {datetime.now().strftime('%Y-%m-%d %H:%M')}",
        f"対象記事: {len(df)} 本",
        f"評価指標: {metric}",
        "",
        "## 全体の数字",
        "",
        f"- 累計PV: {int(df['pv'].sum()):,}",
        f"- 累計スキ: {int(df['likes'].sum()):,}",
        f"- 1本あたり平均スキ: {df['likes'].mean():.1f}",
        f"- 有料記事: {int(df['is_paid'].sum())} 本",
        "",
        "## タイトルの特徴別",
        "",
        "| 特徴 | あり | なし | 比率 |",
        "|------|------|------|------|",
    ]

    for column, label in FEATURES.items():
        with_f = df[df[column] == 1][metric]
        without = df[df[column] == 0][metric]

        if len(with_f) < 3 or len(without) < 3:
            continue

        a = with_f.mean()
        b = without.mean()
        ratio = a / b if b else 0

        lines.append(
            f"| {label} | {a:.2f}（{len(with_f)}本） | "
            f"{b:.2f}（{len(without)}本） | {ratio:.2f} |"
        )

    lines += ["", "## 上位10本", ""]
    for _, row in df.nlargest(10, metric).iterrows():
        lines.append(
            f"- {row[metric]:.2f} / {str(row['title'])[:50]}"
        )

    lines += ["", "## 月別の投稿数", ""]
    monthly = df.groupby("year_month").size()
    for month, count in monthly.tail(12).items():
        lines.append(f"- {month}: {count} 本")

    lines += [
        "",
        "## 次にやること",
        "",
        "1. 上位10本と下位10本を読み比べる",
        "2. 共通点を3つ書き出す",
        "3. 次の5本でそのパターンを試す",
        "4. 3ヶ月後に再集計する",
        "",
        "※ 数字は関連を示すだけで、因果を証明しない",
    ]

    return "\n".join(lines)


def main():
    df = build()
    report = build_report(df)

    with open(OUTPUT, "w", encoding="utf-8") as f:
        f.write(report)

    print(report)
    print(f"\n{OUTPUT} に保存しました")


if __name__ == "__main__":
    main()