kgrowth：GSCとアクセスログを統合した週次グロース自動化パイプライン

def _sign_rs256(private_key: str, signing_input: str) -> bytes:
    try:
        from cryptography.hazmat.primitives import hashes, serialization
        from cryptography.hazmat.primitives.asymmetric import padding
    except Exception:
        # cryptographyがない場合はopenssl CLIにフォールバック
        with tempfile.NamedTemporaryFile("w", delete=True) as key_file:
            key_file.write(private_key)
            key_file.flush()
            proc = subprocess.run(
                ["openssl", "dgst", "-sha256", "-sign", key_file.name],
                input=signing_input.encode("utf-8"),
                stdout=subprocess.PIPE,
            )
        return proc.stdout
    key = serialization.load_pem_private_key(private_key.encode("utf-8"), password=None)
    return key.sign(signing_input.encode("utf-8"), padding.PKCS1v15(), hashes.SHA256())

# Apache Combined Log
LOG_RE = re.compile(
    r'(?P<ip>\S+) \S+ \S+ \[(?P<time>[^\]]+)\] '
    r'"(?P<method>\S+) (?P<path>[^"]*?) (?P<proto>[^"]*)" '
    r'(?P<status>\d{3}) (?P<bytes>\S+) "(?P<referer>[^"]*)" "(?P<ua>[^"]*)"'
)

# simpletrack形式（パイプ区切り）
# 2026-01-01 12:00:00|1.2.3.4|https://example.com/path|https://referer.com|UA文字列
if "|" in line and line[:4].isdigit():
    parts = line.rstrip("\n").split("|", 4)

BOT_WORDS = (
    "bot", "crawler", "spider", "curl", "wget", "python",
    "googlebot", "bingbot", "gptbot", "claudebot", "ccbot",
    ...
)

def is_bot_ua(ua: str) -> bool:
    ua = ua.lower().strip()
    if not ua:
        return True
    return any(word in ua for word in BOT_WORDS)

def _track_go_click(parsed, referer, ...):
    quality = first("click_quality").lower()
    likely_human = quality == "likely_human"
    if not quality:
        likely_human = bool(referer)  # refererがあれば人間と判定

def expected_ctr(pos: float) -> float:
    if pos <= 1:  return 0.28
    if pos <= 2:  return 0.15
    if pos <= 3:  return 0.10
    if pos <= 5:  return 0.06
    if pos <= 10: return 0.025
    return 0.005

# 検索順位10位以内で、期待CTRの50%未満ならタイトル改善候補
if entry["pos"] <= 10 and entry["imp"] >= title_min and entry["ctr"] < expected_ctr(entry["pos"]) * 0.5:
    title_fixes.append(...)

# 11〜30位で表示回数があれば、コンテンツ強化候補
if 10 < entry["pos"] <= 30 and entry["imp"] >= boost_min:
    boost_queries.append(...)

def tokenize(query: str) -> list[str]:
    parts = re.split(r"[\s　]+", query.strip())
    tokens = []
    for part in parts:
        if len(part) >= 2:
            tokens.append(part)
        if re.match(r"^[A-Za-z0-9-]{4,}$", part):
            tokens.append(part.upper())  # 英数字は大文字正規化
    return tokens

def _job_id(kind: str, payload: dict) -> str:
    raw = json.dumps({"kind": kind, "payload": payload}, ensure_ascii=False, sort_keys=True)
    return hashlib.sha1(raw.encode("utf-8")).hexdigest()[:16]