Search Consoleの書き出しを2期間で集計し、直す記事を元の行で確かめて1件選ぶ

import csv
import json
import math
import sys
from collections import defaultdict
from datetime import date

COLUMNS = ["period", "page", "query", "clicks", "impressions", "position"]


def to_count(value, line):
    """空欄は取得できなかった値（None）、確認済みのゼロは0として読む。"""
    if value == "":
        return None
    if value.isascii() and value.isdigit():
        return int(value)
    raise ValueError(f"非負整数または空欄が必要: CSV {line}行")


def to_position(value, line):
    """掲載順位は1以上の数値。空欄は取得できなかった値（None）。"""
    if value == "":
        return None
    try:
        position = float(value)
    except ValueError:
        position = math.nan
    if not math.isfinite(position) or position < 1:
        raise ValueError(f"掲載順位は1以上の数値または空欄: CSV {line}行")
    return position


def aggregate(csv_path, context_path):
    with open(context_path, encoding="utf-8") as f:
        context = json.load(f)
    if set(context) != {"before", "after"}:
        raise ValueError("期間はbeforeとafterの2つが必要")
    spans = []
    for period in ("before", "after"):
        c = context[period]
        start, end = date.fromisoformat(c["start"]), date.fromisoformat(c["end"])
        if end < start or c["finalized"] is not True:
            raise ValueError("期間の順序または確定状態が不正")
        spans.append((start, end))
    if ((spans[0][1] - spans[0][0]).days !=
            (spans[1][1] - spans[1][0]).days or spans[0][1] >= spans[1][0]):
        raise ValueError("期間の長さが不一致、または重複・逆順")
    for key in ("search_type", "country", "device", "grain"):
        if not context["before"][key] or context["before"][key] != context["after"][key]:
            raise ValueError("比較条件が不一致: " + key)
    if context["before"]["grain"] != "query-page":
        raise ValueError("query-page形式が必要")

    grouped = defaultdict(lambda: defaultdict(dict))
    with open(csv_path, encoding="utf-8-sig", newline="") as f:
        reader = csv.DictReader(f)
        if reader.fieldnames != COLUMNS:
            raise ValueError("CSV列が不正")
        for line, row in enumerate(reader, 2):
            if None in row.values():
                raise ValueError(f"列が不足: CSV {line}行")
            p, url, query = row["period"], row["page"], row["query"]
            if p not in context or not url.startswith("https://") or not query:
                raise ValueError(f"入力不正: CSV {line}行")
            if query in grouped[url][p]:
                raise ValueError(f"query-page重複: CSV {line}行")
            clicks = to_count(row["clicks"], line)
            impressions = to_count(row["impressions"], line)
            if clicks is not None and impressions is not None and clicks > impressions:
                raise ValueError(f"クリック数が表示回数を超過: CSV {line}行")
            grouped[url][p][query] = {"clicks": clicks, "impressions": impressions,
                                     "position": to_position(row["position"], line),
                                     "line": line}

    if not grouped:
        raise ValueError("データ行がない")
    output = []
    for url, periods in sorted(grouped.items()):
        reasons, totals = [], {}
        if set(periods.get("before", {})) != set(periods.get("after", {})):
            reasons.append("観測クエリ集合が不一致（欠落はゼロにしない）")
        for p in ("before", "after"):
            rows = list(periods.get(p, {}).values())
            blank = [str(r["line"]) for r in rows
                     if r["clicks"] is None or r["impressions"] is None
                     or (r["impressions"] > 0 and r["position"] is None)]
            if not rows or blank:
                totals[p] = {"clicks": None, "impressions": None,
                             "ctr_pct": None, "position": None}
                reasons.append(p + (": 行なし" if not rows
                                    else ": 空欄あり（CSV " + "・".join(blank) + "行）"))
                continue
            clicks = sum(r["clicks"] for r in rows)
            impressions = sum(r["impressions"] for r in rows)
            if impressions:
                ctr = 100 * clicks / impressions
                # 行ごとの掲載順位を表示回数で重み付けした近似値
                position = sum(r["position"] * r["impressions"]
                               for r in rows if r["impressions"]) / impressions
            else:
                ctr = position = None
                reasons.append(p + ": 表示回数ゼロでCTR・順位を計算できない")
            totals[p] = {"clicks": clicks, "impressions": impressions,
                         "ctr_pct": ctr, "position": position}
        comparable = not reasons
        delta = {}
        if comparable:
            before, after = totals["before"], totals["after"]
            for k in ("clicks", "impressions"):
                delta[k] = after[k] - before[k]
            delta["ctr_pp"] = after["ctr_pct"] - before["ctr_pct"]
            delta["position"] = after["position"] - before["position"]  # 正は下位へ
        output.append({"page": url, "totals": totals, "delta": delta,
                       "comparable": comparable, "hold_reasons": reasons,
                       "source_rows": dict(periods)})
    return output


if __name__ == "__main__":
    try:
        result = aggregate(sys.argv[1], sys.argv[2])
        print(json.dumps(result, ensure_ascii=False, indent=2))
    except (ValueError, KeyError, IndexError) as e:
        sys.exit("入力を確認してください: " + str(e))

"""Search Consoleの書き出し（ZIP）を、aggregate.pyの入力行に写す。

期間比較なし・ページ1件で絞り込んで「CSVをダウンロード」した前後2つのZIPを使う。
"""
import csv
import io
import sys
import zipfile

QUERY_COLUMNS = ["上位のクエリ", "クリック数", "表示回数", "CTR", "掲載順位"]


def read_csv(z, name):
    with z.open(name) as f:
        return list(csv.reader(io.TextIOWrapper(f, encoding="utf-8-sig", newline="")))


def convert(zip_path, period):
    with zipfile.ZipFile(zip_path) as z:
        filters = dict(read_csv(z, "フィルタ.csv")[1:])
        rows = read_csv(z, "クエリ.csv")
    page = filters.get("ページ", "")
    if not page.startswith("https://"):
        raise ValueError(f"{zip_path}: ページ1件で絞り込んだ書き出しが必要")
    if rows[0] != QUERY_COLUMNS:
        raise ValueError(f"{zip_path}: クエリ.csvの列が想定と違う（期間比較の書き出しは使わない）")
    out = []
    for query, clicks, impressions, _ctr, position in rows[1:]:
        if impressions == "0":
            position = ""  # 表示0の行には順位がない
        out.append([period, page, query, clicks, impressions, position])
    return filters, out


if __name__ == "__main__":
    if len(sys.argv) != 3:
        sys.exit("使い方: python3 gsc_to_input.py 前期間.zip 後期間.zip > input.csv")
    try:
        (fb, before), (fa, after) = convert(sys.argv[1], "before"), convert(sys.argv[2], "after")
        if {k: v for k, v in fb.items() if k != "日付"} != {k: v for k, v in fa.items() if k != "日付"}:
            raise ValueError("前後で日付以外のフィルタが違う")
    except (ValueError, KeyError, OSError, zipfile.BadZipFile) as e:
        sys.exit("書き出しを確認してください: " + str(e))
    for name, f, rows in (("before", fb, before), ("after", fa, after)):
        print(f"{name}: {f.get('日付')} / 検索タイプ {f.get('検索タイプ')} / クエリ {len(rows)}行", file=sys.stderr)
    writer = csv.writer(sys.stdout, lineterminator="\n")
    writer.writerow(["period", "page", "query", "clicks", "impressions", "position"])
    writer.writerows(before + after)