"""下降トレンドの早期警告（黄色ランプ判定）。
蓄積した Search Console データから「検索での露出が縮み続けていないか」を機械的に判定する。
点灯は「すぐ何かをする」合図ではなく、「次の見直しの議題に上げる」ための目印として使う。
判定指標:
(1) 表示回数の四半期(91日)合計の成長率が 2四半期連続でマイナス
(2) 「四半期末の直近7日に検索表示があったページ数」が 前々→前→現 と連続で減少
仕様:
- 四半期の区切りは DB 内の GSC データの最新日を基準にする（データ遅延を自動で吸収）
- page_path は /index.html をディレクトリに統合して数える
- データが3四半期分に満たない指標は「判定不能（データ蓄積中）」
使い方:
（プロジェクトのルートで実行する）
python -X utf8 scripts/analytics/yellow_lamp.py # config.yaml の analytics.database を読み、全サイト判定
python -X utf8 scripts/analytics/yellow_lamp.py --site site_a # 1サイトだけ
python -X utf8 scripts/analytics/yellow_lamp.py --db data/analytics.db # DBファイルを直接指定
from yellow_lamp import connect, judge_site # 他のスクリプトからの import も可
"""
from __future__ import annotations
import argparse
import os
import sqlite3
from datetime import date, timedelta
from pathlib import Path
QUARTER_DAYS = 91 # 四半期 = 91日ブロックで近似
# /index.html をディレクトリに統合する SQL 式
NORM = """CASE WHEN page_path='/index.html' THEN '/'
WHEN page_path LIKE '%/index.html' THEN substr(page_path, 1, length(page_path)-11)
ELSE page_path END"""
def connect(db_path: str) -> sqlite3.Connection:
"""DB を読み取り専用で開く。"""
uri = Path(db_path).resolve().as_uri() + "?mode=ro"
return sqlite3.connect(uri, uri=True)
def _quarter_ranges(anchor: date) -> list[tuple[str, str]]:
"""anchor(=GSC最新日)から遡る3つの四半期 [(start,end)] を新しい順に返す。"""
ranges = []
end = anchor
for _ in range(3):
start = end - timedelta(days=QUARTER_DAYS - 1)
ranges.append((start.isoformat(), end.isoformat()))
end = start - timedelta(days=1)
return ranges # [Q0(現), Q1(前), Q2(前々)]
def judge_site(site: str, conn: sqlite3.Connection) -> dict:
"""1サイトの判定。dict（lamp / 各指標の詳細）を返す。
lamp: True=点灯 / False=消灯 / None=判定不能
"""
q = conn.execute
row = q(
"SELECT MIN(date), MAX(date) FROM gsc_page_metrics WHERE site=?", (site,)
).fetchone()
if not row or not row[1]:
return {"site": site, "lamp": None, "reason": "GSCデータなし"}
data_min, anchor = date.fromisoformat(row[0]), date.fromisoformat(row[1])
quarters = _quarter_ranges(anchor)
# 3四半期ぶんのデータが揃っているか（Q2の開始がデータ最古日以降か）
enough = date.fromisoformat(quarters[2][0]) >= data_min
# --- 指標(1): 四半期ごとの表示回数合計 → 成長率が2回連続マイナス ---
imps = []
for s, e in quarters:
v = q(
"""SELECT COALESCE(SUM(impressions),0) FROM gsc_page_metrics
WHERE site=? AND date BETWEEN ? AND ?""",
(site, s, e),
).fetchone()[0]
imps.append(int(v))
def growth(cur: int, prev: int):
if prev <= 0:
return None # 分母0は判定不能扱い
return (cur - prev) / prev
g0 = growth(imps[0], imps[1]) # 現 vs 前
g1 = growth(imps[1], imps[2]) # 前 vs 前々
if not enough or g0 is None or g1 is None:
ind1 = {"lit": None, "note": "判定不能（データ蓄積中）"}
else:
ind1 = {"lit": g0 < 0 and g1 < 0}
ind1.update({
"impressions_q": imps,
"growth_recent": round(g0, 4) if g0 is not None else None,
"growth_prior": round(g1, 4) if g1 is not None else None,
})
# --- 指標(2): 各四半期末7日間の「表示があったページ数」→ 2段連続で減少 ---
pages = []
for _, e in quarters:
e_d = date.fromisoformat(e)
s7 = (e_d - timedelta(days=6)).isoformat()
v = q(
f"""SELECT COUNT(DISTINCT {NORM}) FROM gsc_page_metrics
WHERE site=? AND date BETWEEN ? AND ? AND impressions > 0""",
(site, s7, e),
).fetchone()[0]
pages.append(int(v))
# 最も古い7日窓がデータ最古日より前にはみ出す場合は判定不能
oldest_window_start = date.fromisoformat(quarters[2][1]) - timedelta(days=6)
enough2 = oldest_window_start >= data_min
if not enough2:
ind2 = {"lit": None, "note": "判定不能（データ蓄積中）"}
else:
ind2 = {"lit": pages[0] < pages[1] and pages[1] < pages[2]}
ind2["active_pages_q"] = pages # [現, 前, 前々]
lit_flags = [v["lit"] for v in (ind1, ind2)]
lamp = True if any(f is True for f in lit_flags) else (
None if all(f is None for f in lit_flags) else False
)
return {
"site": site,
"anchor": anchor.isoformat(),
"quarters": quarters,
"indicator1": ind1,
"indicator2": ind2,
"lamp": lamp,
}
def format_report(r: dict) -> str:
"""人間が読める形式の判定結果（レポートの冒頭にそのまま貼れる）。"""
if r.get("lamp") is None and "reason" in r:
return f"[{r['site']}] 黄色ランプ: 判定不能（{r['reason']}）"
lines = []
head = {True: "[警告] 黄色ランプ点灯", False: "[OK] 黄色ランプ消灯",
None: "[保留] 判定不能（データ蓄積中）"}[r["lamp"]]
lines.append(f"[{r['site']}] {head}（基準日: {r['anchor']}）")
i1, i2 = r["indicator1"], r["indicator2"]
q_imp = i1["impressions_q"]
g0, g1 = i1.get("growth_recent"), i1.get("growth_prior")
pct = lambda v: f"{v*100:+.1f}%" if v is not None else "—" # noqa: E731
mark = lambda lit: {True: "警告", False: "OK", None: "保留"}[lit] # noqa: E731
lines.append(
f" 指標(1) 表示回数の四半期成長: {mark(i1['lit'])} "
f"(前々→前 {pct(g1)}, 前→現 {pct(g0)} / 四半期計 {q_imp[2]:,}→{q_imp[1]:,}→{q_imp[0]:,})"
)
p = i2["active_pages_q"]
lines.append(
f" 指標(2) 検索表示ページ数(各四半期末7日): {mark(i2['lit'])} "
f"({p[2]}→{p[1]}→{p[0]} ページ)"
)
if r["lamp"]:
lines.append(" → 点灯は「すぐ対応」の合図ではありません。次回の見直しの議題に上げてください。")
return "\n".join(lines)
def _resolve_db_path(args) -> str:
"""--db があればそれを、無ければ設定ファイルの analytics.database を使う。"""
if args.db:
return args.db
import common # 設定ファイルを読むときだけ必要（PyYAML に依存）
return common.load_config(args.config)["database"]
def main() -> int:
parser = argparse.ArgumentParser(description="下降トレンドの早期警告（黄色ランプ判定）")
parser.add_argument("--site", default=None, help="サイト指定（省略時は DB 内の全サイト）")
parser.add_argument("--db", default=None, help="SQLite ファイルのパス（省略時は設定ファイルの analytics.database）")
parser.add_argument("--config", default=None, help="設定ファイルのパス（既定: プロジェクトのルートの config.yaml。読むのは analytics: 節）")
args = parser.parse_args()
db_path = _resolve_db_path(args)
if not os.path.exists(db_path):
print(f"[ERROR] DBファイルがありません: {db_path}")
return 2
conn = connect(db_path)
try:
if args.site:
sites = [args.site]
else:
sites = [r[0] for r in conn.execute(
"SELECT DISTINCT site FROM gsc_page_metrics ORDER BY site")]
if not sites:
print("黄色ランプ: 判定不能（データ不足: GSCデータがまだ1件もありません）")
return 0
for s in sites:
print(format_report(judge_site(s, conn)))
print()
finally:
conn.close()
return 0
if __name__ == "__main__":
raise SystemExit(main())
MIT ライセンスで配布しています。