AI（Genie）と探る人気アニメヒットの法則

-- 使用するカタログとスキーマ
USE CATALOG test_uc;
CREATE SCHEMA IF NOT EXISTS anime;
USE SCHEMA anime;

-- 作品メタ情報
CREATE TABLE IF NOT EXISTS anime.items (
  qid STRING COMMENT 'WikidataのID',
  title_ja STRING COMMENT '作品タイトル（日本語）',
  title_en STRING COMMENT '作品タイトル（英語）',
  genre ARRAY COMMENT 'ジャンルのリスト',
  studio ARRAY COMMENT '制作会社のリスト',
  country ARRAY COMMENT '製作国のリスト',
  first_air_year INT COMMENT '初回放送年',
  wikimedia_title STRING COMMENT 'Wikipediaの記事タイトル',
  wikipedia_project STRING COMMENT "'ja.wikipedia' など",
  _ingested_at TIMESTAMP COMMENT 'データ取得日時'
) USING DELTA;

-- 日次ページビュー
CREATE TABLE IF NOT EXISTS anime.pageviews (
  date DATE COMMENT '日付',
  wikimedia_title STRING COMMENT 'Wikipediaの記事タイトル',
  wikipedia_project STRING COMMENT "'ja.wikipedia' など",
  views BIGINT COMMENT '閲覧数',
  _ingested_at TIMESTAMP COMMENT 'データ取得日時'
) USING DELTA
PARTITIONED BY (date);

# %python
"""
Wikidata → Spark DataFrame → Delta（anime.items）
- 対象：アニメシリーズ（Q581714） or そのジャンルを持つ作品
- 条件：ja.wikipedia に記事があるもの
- 出力：test_uc.anime.items（Delta）
"""

import re
import requests
from typing import Any, Dict, List, Optional

from pyspark.sql import Row, functions as F, types as T, Window

# ============================================================
# 0) 定数・環境設定
# ============================================================

# ★必ず自社の連絡先に置き換えてください（Wikimediaポリシー順守）
USER_AGENT = "<クライアント名>/<バージョン> (<連絡先情報>)"

CATALOG = "test_uc"
SCHEMA = "anime"
OUTPUT_TABLE = f"{SCHEMA}.items"

ENDPOINT = "https://query.wikidata.org/sparql"

SPARQL = """
SELECT DISTINCT
  ?item
  (SAMPLE(?jaLabel) AS ?jaLabel)
  (SAMPLE(?enLabel) AS ?enLabel)
  (GROUP_CONCAT(DISTINCT ?genreLabel;   separator="|") AS ?genreLabels)
  (GROUP_CONCAT(DISTINCT ?studioLabel;  separator="|") AS ?studioLabels)
  (GROUP_CONCAT(DISTINCT ?countryLabel; separator="|") AS ?countryLabels)
  (MIN(?pubDate) AS ?pubFirstDate)
  ?article ?wiki ?title
WHERE {
  # instance/subclass of: animated series
  { ?item wdt:P31/wdt:P279* wd:Q581714 . }
  UNION
  # genre includes: animated series
  { ?item wdt:P136/wdt:P279* wd:Q581714 . }

  OPTIONAL { ?item wdt:P136  ?genre .   ?genre   rdfs:label ?genreLabel   FILTER (lang(?genreLabel)   IN ("ja","en")) }
  OPTIONAL { ?item wdt:P272  ?studio .  ?studio  rdfs:label ?studioLabel  FILTER (lang(?studioLabel)  IN ("ja","en")) }
  OPTIONAL { ?item wdt:P495  ?country . ?country rdfs:label ?countryLabel FILTER (lang(?countryLabel) IN ("ja","en")) }
  OPTIONAL { ?item wdt:P577  ?pubDate . }

  # sitelink（日本語版のみ）
  ?article schema:about ?item ;
           schema:isPartOf ?wiki ;
           schema:name ?title .
  FILTER (?wiki IN (<https://ja.wikipedia.org/>))

  OPTIONAL { ?item rdfs:label ?jaLabel FILTER (lang(?jaLabel) = "ja") }
  OPTIONAL { ?item rdfs:label ?enLabel FILTER (lang(?enLabel) = "en") }
}
GROUP BY ?item ?article ?wiki ?title
LIMIT 5000
"""

spark.sql(f"USE CATALOG {CATALOG}")
spark.sql(f"USE SCHEMA {SCHEMA}")

# ============================================================
# 1) ユーティリティ
# ============================================================

def jval(binding: Dict[str, Any], key: str) -> Optional[str]:
    """SPARQLの1件（binding）から値を安全に取得。"""
    v = binding.get(key)
    return v.get("value") if isinstance(v, dict) and "value" in v else None

def split_pipe_unique(pipe_str: Optional[str]) -> List[str]:
    """'A|B|B|C' → ['A','B','C']（順序維持・重複排除）"""
    if not pipe_str:
        return []
    out, seen = [], set()
    for token in (p.strip() for p in pipe_str.split("|")):
        if token and token not in seen:
            seen.add(token)
            out.append(token)
    return out

def parse_first_year(iso_date: Optional[str]) -> Optional[int]:
    """'YYYY-MM-DD' などから年だけ抽出（不正なら None）。"""
    if not iso_date:
        return None
    m = re.match(r"^(\d{4})", iso_date)
    return int(m.group(1)) if m else None

def qid_from_uri(uri: Optional[str]) -> Optional[str]:
    """'http://www.wikidata.org/entity/Q123' → 'Q123'"""
    if not uri:
        return None
    return uri.rsplit("/", 1)[-1] if uri.startswith("http") else uri

# ============================================================
# 2) SPARQL 実行
# ============================================================

session = requests.Session()
resp = session.get(
    ENDPOINT,
    params={"query": SPARQL, "format": "json"},
    headers={"User-Agent": USER_AGENT},
    timeout=60,
)
resp.raise_for_status()
bindings = resp.json()["results"]["bindings"]

# ============================================================
# 3) JSON → Row 化
# ============================================================

rows: List[Row] = []
for b in bindings:
    qid = qid_from_uri(jval(b, "item"))

    # 日本語版のみを対象（英語版は除外）
    project = "ja.wikipedia"
    title = jval(b, "title") or ""  # ja の記事タイトル

    # ラベルと属性
    title_ja = jval(b, "jaLabel")
    title_en = jval(b, "enLabel")
    genres    = split_pipe_unique(jval(b, "genreLabels"))
    studios   = split_pipe_unique(jval(b, "studioLabels"))
    countries = split_pipe_unique(jval(b, "countryLabels"))
    first_air_year = parse_first_year(jval(b, "pubFirstDate"))

    rows.append(Row(
        qid=qid,
        title_ja=title_ja,
        title_en=title_en,
        genre=genres,
        studio=studios,
        country=countries,
        first_air_year=first_air_year,
        wikimedia_title=title,     # APIに渡す元タイトル（ja）
        wikipedia_project=project, # 'ja.wikipedia' 固定
    ))

# ============================================================
# 4) DataFrame 化 & 重複除去・付加情報
# ============================================================

schema = T.StructType([
    T.StructField("qid", T.StringType()),
    T.StructField("title_ja", T.StringType()),
    T.StructField("title_en", T.StringType()),
    T.StructField("genre", T.ArrayType(T.StringType())),
    T.StructField("studio", T.ArrayType(T.StringType())),
    T.StructField("country", T.ArrayType(T.StringType())),
    T.StructField("first_air_year", T.IntegerType()),
    T.StructField("wikimedia_title", T.StringType()),
    T.StructField("wikipedia_project", T.StringType()),
])

df = spark.createDataFrame(rows, schema)

# 同一 qid の多重行が返る可能性に備え、代表1行へ（ja→enの順で安定）
w = Window.partitionBy("qid").orderBy(
    F.coalesce(F.col("title_ja"), F.col("title_en")).asc_nulls_last()
)
df_dedup = (
    df.withColumn("rn", F.row_number().over(w))
      .filter("rn = 1")
      .drop("rn")
)

# 監査用タイムスタンプを付与
df_final = df_dedup.withColumn("_ingested_at", F.current_timestamp())

# ============================================================
# 5) Delta へ保存
# ============================================================

(df_final
 .write
 .format("delta")
 .mode("overwrite")               # 既存構造を置き換え
 .option("overwriteSchema", "true")
 .saveAsTable(OUTPUT_TABLE))

-- %sql
USE CATALOG test_uc;
CREATE SCHEMA IF NOT EXISTS anime;
USE SCHEMA anime;

CREATE TABLE IF NOT EXISTS anime.items(
  qid STRING,
  title_ja STRING,
  title_en STRING,
  genre ARRAY,
  studio ARRAY,
  first_air_year INT,
  country ARRAY,
  wikimedia_title STRING,
  wikipedia_project STRING,
  _ingested_at TIMESTAMP
)
USING DELTA
TBLPROPERTIES (
  delta.enableChangeDataFeed = true
);

CREATE TABLE IF NOT EXISTS anime.pageviews (
  date DATE,
  wikimedia_title STRING,
  wikipedia_project STRING,
  views BIGINT,
  _ingested_at TIMESTAMP
) USING DELTA
PARTITIONED BY (date);

# %python
"""
Wikipedia Pageviews（ja）を 直近365日 分取得し、一時ビュー 'pv_stage_last365' に格納
- 取得対象: test_uc.anime.items の (wikimedia_title, wikipedia_project='ja.wikipedia')
- API: /metrics/pageviews/per-article/{project}.org/all-access/user/{title}/daily/{start}/{end}
"""

import time
import urllib.parse
import requests
from datetime import date, timedelta
from typing import List

from pyspark.sql import Row, functions as F, types as T

# =========================
# 0) 定数・環境
# =========================
spark.sql("USE CATALOG test_uc")
spark.sql("USE SCHEMA anime")

# ★Wikimediaポリシー順守: 自社連絡先を必ず設定
USER_AGENT = "<クライアント名>/<バージョン> (<連絡先情報>)"

BASE_URL     = "https://wikimedia.org/api/rest_v1/metrics/pageviews/per-article"
PAUSE_EVERY  = 20      # 何リクエストごとにスリープするか
SLEEP_SEC    = 1.5     # 429回避。必要に応じて増やす
BATCH_LIMIT  = 3000   

end = date.today()
start = end - timedelta(days=365)
start_str, end_str = start.strftime("%Y%m%d"), end.strftime("%Y%m%d")

# =========================
# 1) 取得対象の抽出（ja のみ）
# =========================
targets: List[Row] = (
    spark.table("anime.items")
         .select("wikimedia_title", "wikipedia_project")
         .where("wikipedia_project = 'ja.wikipedia' AND wikimedia_title IS NOT NULL")
         .distinct()
         .limit(BATCH_LIMIT)
         .collect()
)

print("TARGET COUNT:", len(targets))

# =========================
# 2) API コール & バッファ
# =========================
rows, ok, not_found, errors = [], 0, 0, 0

for i, r in enumerate(targets, start=1):
    title = r["wikimedia_title"].replace(" ", "_")
    article = urllib.parse.quote(title, safe="_")
    project_org = r["wikipedia_project"] + ".org"       # 'ja.wikipedia.org'
    url = f"{BASE_URL}/{project_org}/all-access/user/{article}/daily/{start_str}/{end_str}"

    try:
        resp = requests.get(url, headers={"User-Agent": USER_AGENT}, timeout=30)
        if resp.status_code == 404:
            not_found += 1
            continue
        resp.raise_for_status()

        for it in resp.json().get("items", []):
            rows.append(Row(
                date=it["timestamp"][:8],
                wikimedia_title=r["wikimedia_title"],
                wikipedia_project=r["wikipedia_project"],
                views=int(it["views"])
            ))
        ok += 1

    except Exception as e:
        errors += 1
        print("ERR:", r["wikimedia_title"], e)

    if i % PAUSE_EVERY == 0:
        time.sleep(SLEEP_SEC)

print(f"RESULT: ok={ok}, not_found={not_found}, errors={errors}, rows={len(rows)}")

# =========================
# 3) Spark DataFrame へ → 一時ビュー
# =========================
schema = T.StructType([
    T.StructField("date", T.StringType()),
    T.StructField("wikimedia_title", T.StringType()),
    T.StructField("wikipedia_project", T.StringType()),
    T.StructField("views", T.LongType()),
])

pv_df = spark.createDataFrame(rows, schema) if rows else spark.createDataFrame([], schema)

pv_df = (
    pv_df
      .withColumn("date", F.to_date("date", "yyyyMMdd"))
      .withColumn("_ingested_at", F.current_timestamp())
)

pv_df.createOrReplaceTempView("pv_stage_last365")

-- %sql
USE CATALOG test_uc;
USE SCHEMA anime;

MERGE INTO anime.pageviews AS t
USING (
  SELECT DISTINCT date, wikimedia_title, wikipedia_project, views, _ingested_at
  FROM pv_stage_last365
) AS s
ON  t.date = s.date
AND t.wikimedia_title = s.wikimedia_title
AND t.wikipedia_project = s.wikipedia_project
WHEN MATCHED THEN UPDATE SET
  t.views = s.views,
  t._ingested_at = s._ingested_at
WHEN NOT MATCHED THEN INSERT * ;

-- %sql
-- 直近30日平均PV
CREATE OR REPLACE VIEW anime.v_anime_pv30 AS
SELECT i.qid,
       COALESCE(i.title_ja, i.title_en) AS title,
       i.genre, i.studio, i.first_air_year,
       AVG(p.views) AS avg_views_30d
FROM anime.items i
JOIN anime.pageviews p
  ON p.wikimedia_title = i.wikimedia_title
 AND p.wikipedia_project = i.wikipedia_project
WHERE p.date >= date_sub(current_date(), 30)
GROUP BY ALL;

-- 直近12か月の月次平均PV（可視化用）
CREATE OR REPLACE VIEW anime.v_anime_pv_monthly_last12 AS
SELECT COALESCE(i.title_ja, i.title_en) AS title,
       DATE_TRUNC('month', p.date) AS ym,
       AVG(p.views) AS avg_views_month,
       i.genre, i.studio, i.first_air_year
FROM anime.items i
JOIN anime.pageviews p
  ON p.wikimedia_title = i.wikimedia_title
 AND p.wikipedia_project = i.wikipedia_project
WHERE p.date >= add_months(current_date(), -12)
GROUP BY ALL;