1734 lines
78 KiB
Python
1734 lines
78 KiB
Python
"""统一直播统计 SQLite 存储层。
|
|
|
|
所有写入经由单个 asyncio.Queue 消费协程串行执行。业务侧 record_* 方法均为
|
|
best-effort：队列未启动、已关闭、数据不合法或数据库暂时不可用时不会向主业务抛出异常。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import ctypes
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import os
|
|
import sqlite3
|
|
import time
|
|
from collections.abc import Mapping, Sequence
|
|
from datetime import UTC, datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib.parse import urlsplit, urlunsplit
|
|
|
|
try:
|
|
from .core.runtime_paths import APP_ROOT
|
|
except ImportError: # 允许作为独立模块加载
|
|
APP_ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
LOGGER = logging.getLogger(__name__)
|
|
BEIJING_TZ = timezone(timedelta(hours=8), name="Asia/Shanghai")
|
|
DEFAULT_DATABASE_PATH = APP_ROOT / "data" / "statistics.sqlite3"
|
|
_SCHEMA_VERSION = 2
|
|
_STOP = object()
|
|
_FLUSH = object()
|
|
|
|
_LIFECYCLE_KEYS = {
|
|
"login_sessions": ("login_session_id", "started_at_utc", "business_date"),
|
|
"group_runs": ("group_run_id", "started_at_utc", "business_date"),
|
|
"song_requests": ("request_id", "requested_at_utc", "business_date"),
|
|
"playback_sessions": ("playback_id", "started_at_utc", "business_date"),
|
|
"broadcast_requests": ("request_id", "requested_at_utc", "business_date"),
|
|
"tts_requests": ("request_id", "requested_at_utc", "business_date"),
|
|
"bilibili_connections": ("connection_id", "connected_at_utc", "business_date"),
|
|
}
|
|
|
|
_SENSITIVE_PARTS = (
|
|
"cookie",
|
|
"token",
|
|
"authorization",
|
|
"api_key",
|
|
"apikey",
|
|
"access_key",
|
|
"secret",
|
|
"password",
|
|
"passwd",
|
|
"music_u",
|
|
"session_key",
|
|
"csrf",
|
|
"buvid",
|
|
)
|
|
_AUDIO_URL_PARTS = ("audio_url", "temp_url", "temporary_url", "voice_url", "tts_url")
|
|
|
|
|
|
def utc_now() -> str:
|
|
"""返回可排序的 UTC ISO-8601 时间。"""
|
|
return datetime.now(UTC).isoformat(timespec="milliseconds").replace("+00:00", "Z")
|
|
|
|
|
|
def business_date(value: datetime | str | None = None) -> str:
|
|
"""按北京时间计算业务日；无时区输入按 UTC 处理。"""
|
|
if isinstance(value, str):
|
|
try:
|
|
current = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
except ValueError:
|
|
current = datetime.now(UTC)
|
|
else:
|
|
current = value or datetime.now(UTC)
|
|
if current.tzinfo is None:
|
|
current = current.replace(tzinfo=UTC)
|
|
return (current.astimezone(BEIJING_TZ) - timedelta(hours=4)).date().isoformat()
|
|
|
|
|
|
def _safe_url(value: str) -> str:
|
|
"""仅保留 URL 的来源与路径，不保存查询串、片段和用户信息。"""
|
|
try:
|
|
parsed = urlsplit(value)
|
|
except ValueError:
|
|
return "[REDACTED_URL]"
|
|
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
|
|
return "[REDACTED_URL]"
|
|
hostname = parsed.hostname or ""
|
|
port = f":{parsed.port}" if parsed.port else ""
|
|
return urlunsplit((parsed.scheme, f"{hostname}{port}", parsed.path, "", ""))
|
|
|
|
|
|
def redact_payload(value: Any, key: str = "") -> Any:
|
|
"""递归复制并脱敏可序列化 payload。"""
|
|
lowered = key.casefold()
|
|
if any(part in lowered for part in _SENSITIVE_PARTS):
|
|
return "[REDACTED]"
|
|
if value is None or isinstance(value, (bool, int, float)):
|
|
return value
|
|
if isinstance(value, str):
|
|
if any(part in lowered for part in _AUDIO_URL_PARTS):
|
|
return "[REDACTED_AUDIO_URL]"
|
|
if lowered.endswith("url") and ("?" in value or "#" in value):
|
|
return _safe_url(value)
|
|
return value
|
|
if isinstance(value, bytes):
|
|
return f"[BYTES:{len(value)}]"
|
|
if isinstance(value, Mapping):
|
|
return {str(item_key): redact_payload(item_value, str(item_key)) for item_key, item_value in value.items()}
|
|
if isinstance(value, Sequence):
|
|
return [redact_payload(item, key) for item in value]
|
|
return str(value)
|
|
|
|
|
|
def _json(value: Any) -> str:
|
|
return json.dumps(redact_payload(value), ensure_ascii=False, separators=(",", ":"), default=str)
|
|
|
|
|
|
def _utc_iso(value: datetime | str) -> str:
|
|
if isinstance(value, str):
|
|
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
else:
|
|
parsed = value
|
|
if parsed.tzinfo is None:
|
|
parsed = parsed.replace(tzinfo=UTC)
|
|
return parsed.astimezone(UTC).isoformat(timespec="milliseconds").replace("+00:00", "Z")
|
|
|
|
|
|
def _clean_fields(fields: Mapping[str, Any]) -> dict[str, Any]:
|
|
clean = redact_payload(dict(fields))
|
|
assert isinstance(clean, dict)
|
|
for name, value in tuple(clean.items()):
|
|
if name.endswith("_at_utc") and value is not None:
|
|
clean[name] = _utc_iso(value)
|
|
elif isinstance(value, (dict, list, tuple)):
|
|
clean[name] = _json(value)
|
|
elif isinstance(value, bool):
|
|
clean[name] = int(value)
|
|
return clean
|
|
|
|
|
|
_SCHEMA = """
|
|
CREATE TABLE IF NOT EXISTS schema_migrations (
|
|
version INTEGER PRIMARY KEY,
|
|
applied_at_utc TEXT NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS process_sessions (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
session_id TEXT NOT NULL UNIQUE,
|
|
started_at_utc TEXT NOT NULL,
|
|
ended_at_utc TEXT,
|
|
business_date TEXT NOT NULL,
|
|
pid INTEGER,
|
|
host TEXT,
|
|
app_version TEXT,
|
|
exit_reason TEXT,
|
|
metadata_json TEXT NOT NULL DEFAULT '{}'
|
|
);
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
platform TEXT NOT NULL,
|
|
platform_user_id TEXT NOT NULL,
|
|
display_name TEXT,
|
|
avatar_url TEXT,
|
|
user_level INTEGER,
|
|
is_admin INTEGER NOT NULL DEFAULT 0,
|
|
first_seen_at_utc TEXT NOT NULL,
|
|
last_seen_at_utc TEXT NOT NULL,
|
|
snapshot_json TEXT NOT NULL DEFAULT '{}',
|
|
points INTEGER NOT NULL DEFAULT 0,
|
|
UNIQUE(platform, platform_user_id)
|
|
);
|
|
CREATE TABLE IF NOT EXISTS signin_events (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
event_id TEXT UNIQUE,
|
|
user_id INTEGER REFERENCES users(id),
|
|
platform TEXT,
|
|
platform_user_id TEXT,
|
|
signed_at_utc TEXT NOT NULL,
|
|
business_date TEXT NOT NULL,
|
|
points_awarded INTEGER NOT NULL DEFAULT 0,
|
|
streak_days INTEGER,
|
|
status TEXT,
|
|
payload_json TEXT NOT NULL DEFAULT '{}'
|
|
);
|
|
CREATE TABLE IF NOT EXISTS point_transactions (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
transaction_id TEXT NOT NULL UNIQUE,
|
|
user_id INTEGER REFERENCES users(id),
|
|
platform TEXT,
|
|
platform_user_id TEXT,
|
|
occurred_at_utc TEXT NOT NULL,
|
|
business_date TEXT NOT NULL,
|
|
amount INTEGER NOT NULL,
|
|
balance_after INTEGER,
|
|
reason TEXT,
|
|
reference_type TEXT,
|
|
reference_id TEXT,
|
|
payload_json TEXT NOT NULL DEFAULT '{}'
|
|
);
|
|
CREATE TABLE IF NOT EXISTS queue_events (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
event_id TEXT UNIQUE,
|
|
occurred_at_utc TEXT NOT NULL,
|
|
business_date TEXT NOT NULL,
|
|
queue_name TEXT NOT NULL,
|
|
action TEXT NOT NULL,
|
|
item_id TEXT,
|
|
user_id INTEGER REFERENCES users(id),
|
|
position INTEGER,
|
|
queue_size INTEGER,
|
|
wait_ms INTEGER,
|
|
payload_json TEXT NOT NULL DEFAULT '{}'
|
|
);
|
|
CREATE TABLE IF NOT EXISTS login_sessions (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
login_session_id TEXT NOT NULL UNIQUE,
|
|
user_id INTEGER REFERENCES users(id),
|
|
platform TEXT,
|
|
platform_user_id TEXT,
|
|
started_at_utc TEXT NOT NULL,
|
|
ended_at_utc TEXT,
|
|
business_date TEXT NOT NULL,
|
|
status TEXT,
|
|
client_kind TEXT,
|
|
metadata_json TEXT NOT NULL DEFAULT '{}'
|
|
);
|
|
CREATE TABLE IF NOT EXISTS group_runs (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
group_run_id TEXT NOT NULL UNIQUE,
|
|
group_name TEXT,
|
|
started_at_utc TEXT NOT NULL,
|
|
ended_at_utc TEXT,
|
|
business_date TEXT NOT NULL,
|
|
status TEXT,
|
|
requested_count INTEGER,
|
|
completed_count INTEGER,
|
|
failed_count INTEGER,
|
|
payload_json TEXT NOT NULL DEFAULT '{}'
|
|
);
|
|
CREATE TABLE IF NOT EXISTS song_requests (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
request_id TEXT NOT NULL UNIQUE,
|
|
user_id INTEGER REFERENCES users(id),
|
|
platform TEXT,
|
|
platform_user_id TEXT,
|
|
requested_at_utc TEXT NOT NULL,
|
|
business_date TEXT NOT NULL,
|
|
song_id TEXT,
|
|
song_name TEXT,
|
|
artist TEXT,
|
|
source TEXT,
|
|
status TEXT,
|
|
points_cost INTEGER NOT NULL DEFAULT 0,
|
|
queue_position INTEGER,
|
|
payload_json TEXT NOT NULL DEFAULT '{}'
|
|
);
|
|
CREATE TABLE IF NOT EXISTS playback_sessions (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
playback_id TEXT NOT NULL UNIQUE,
|
|
request_id TEXT REFERENCES song_requests(request_id),
|
|
song_id TEXT,
|
|
song_name TEXT,
|
|
started_at_utc TEXT NOT NULL,
|
|
ended_at_utc TEXT,
|
|
business_date TEXT NOT NULL,
|
|
status TEXT,
|
|
duration_ms INTEGER,
|
|
played_ms INTEGER,
|
|
stop_reason TEXT,
|
|
payload_json TEXT NOT NULL DEFAULT '{}'
|
|
);
|
|
CREATE TABLE IF NOT EXISTS gift_events (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
event_id TEXT NOT NULL UNIQUE,
|
|
user_id INTEGER REFERENCES users(id),
|
|
platform TEXT,
|
|
platform_user_id TEXT,
|
|
occurred_at_utc TEXT NOT NULL,
|
|
business_date TEXT NOT NULL,
|
|
gift_id TEXT,
|
|
gift_name TEXT,
|
|
quantity INTEGER NOT NULL DEFAULT 1,
|
|
unit_value REAL NOT NULL DEFAULT 0,
|
|
total_value REAL NOT NULL DEFAULT 0,
|
|
currency TEXT,
|
|
combo_id TEXT,
|
|
payload_json TEXT NOT NULL DEFAULT '{}'
|
|
);
|
|
CREATE TABLE IF NOT EXISTS gift_aggregates (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
business_date TEXT NOT NULL,
|
|
platform TEXT NOT NULL DEFAULT '',
|
|
platform_user_id TEXT NOT NULL DEFAULT '',
|
|
gift_id TEXT NOT NULL DEFAULT '',
|
|
gift_name TEXT,
|
|
quantity INTEGER NOT NULL DEFAULT 0,
|
|
total_value REAL NOT NULL DEFAULT 0,
|
|
updated_at_utc TEXT NOT NULL,
|
|
UNIQUE(business_date, platform, platform_user_id, gift_id)
|
|
);
|
|
CREATE TABLE IF NOT EXISTS broadcast_requests (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
request_id TEXT NOT NULL UNIQUE,
|
|
user_id INTEGER REFERENCES users(id),
|
|
requested_at_utc TEXT NOT NULL,
|
|
completed_at_utc TEXT,
|
|
business_date TEXT NOT NULL,
|
|
channel TEXT,
|
|
content_length INTEGER,
|
|
status TEXT,
|
|
result_code TEXT,
|
|
payload_json TEXT NOT NULL DEFAULT '{}'
|
|
);
|
|
CREATE TABLE IF NOT EXISTS tts_requests (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
request_id TEXT NOT NULL UNIQUE,
|
|
user_id INTEGER REFERENCES users(id),
|
|
requested_at_utc TEXT NOT NULL,
|
|
completed_at_utc TEXT,
|
|
business_date TEXT NOT NULL,
|
|
voice TEXT,
|
|
text_length INTEGER,
|
|
status TEXT,
|
|
duration_ms INTEGER,
|
|
result_code TEXT,
|
|
payload_json TEXT NOT NULL DEFAULT '{}'
|
|
);
|
|
CREATE TABLE IF NOT EXISTS bilibili_connections (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
connection_id TEXT NOT NULL UNIQUE,
|
|
room_id TEXT,
|
|
connected_at_utc TEXT NOT NULL,
|
|
disconnected_at_utc TEXT,
|
|
business_date TEXT NOT NULL,
|
|
status TEXT,
|
|
reconnect_count INTEGER NOT NULL DEFAULT 0,
|
|
disconnect_reason TEXT,
|
|
payload_json TEXT NOT NULL DEFAULT '{}'
|
|
);
|
|
CREATE TABLE IF NOT EXISTS danmu_events (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
event_id TEXT NOT NULL UNIQUE,
|
|
user_id INTEGER REFERENCES users(id),
|
|
platform TEXT NOT NULL DEFAULT 'bilibili',
|
|
platform_user_id TEXT,
|
|
room_id TEXT,
|
|
occurred_at_utc TEXT NOT NULL,
|
|
business_date TEXT NOT NULL,
|
|
message_type TEXT NOT NULL DEFAULT 'danmu',
|
|
content_length INTEGER,
|
|
command TEXT,
|
|
handled INTEGER NOT NULL DEFAULT 0,
|
|
payload_json TEXT NOT NULL DEFAULT '{}'
|
|
);
|
|
CREATE TABLE IF NOT EXISTS service_state_events (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
event_id TEXT UNIQUE,
|
|
service_name TEXT NOT NULL,
|
|
occurred_at_utc TEXT NOT NULL,
|
|
business_date TEXT NOT NULL,
|
|
old_state TEXT,
|
|
new_state TEXT NOT NULL,
|
|
reason TEXT,
|
|
payload_json TEXT NOT NULL DEFAULT '{}'
|
|
);
|
|
CREATE TABLE IF NOT EXISTS admin_audit_events (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
event_id TEXT NOT NULL UNIQUE,
|
|
occurred_at_utc TEXT NOT NULL,
|
|
business_date TEXT NOT NULL,
|
|
actor TEXT,
|
|
action TEXT NOT NULL,
|
|
target_type TEXT,
|
|
target_id TEXT,
|
|
success INTEGER NOT NULL DEFAULT 1,
|
|
remote_address_hash TEXT,
|
|
payload_json TEXT NOT NULL DEFAULT '{}'
|
|
);
|
|
CREATE TABLE IF NOT EXISTS import_sources (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
source_key TEXT NOT NULL UNIQUE,
|
|
source_type TEXT NOT NULL,
|
|
source_fingerprint TEXT,
|
|
first_imported_at_utc TEXT NOT NULL,
|
|
last_imported_at_utc TEXT NOT NULL,
|
|
status TEXT,
|
|
metadata_json TEXT NOT NULL DEFAULT '{}'
|
|
);
|
|
CREATE TABLE IF NOT EXISTS import_checkpoints (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
source_id INTEGER NOT NULL REFERENCES import_sources(id) ON DELETE CASCADE,
|
|
checkpoint_key TEXT NOT NULL,
|
|
checkpoint_value TEXT,
|
|
updated_at_utc TEXT NOT NULL,
|
|
payload_json TEXT NOT NULL DEFAULT '{}',
|
|
UNIQUE(source_id, checkpoint_key)
|
|
);
|
|
CREATE TABLE IF NOT EXISTS events (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
event_id TEXT NOT NULL UNIQUE,
|
|
event_type TEXT NOT NULL,
|
|
category TEXT,
|
|
occurred_at_utc TEXT NOT NULL,
|
|
business_date TEXT NOT NULL,
|
|
process_session_id TEXT,
|
|
user_id INTEGER REFERENCES users(id),
|
|
platform TEXT,
|
|
platform_user_id TEXT,
|
|
correlation_id TEXT,
|
|
payload_json TEXT NOT NULL DEFAULT '{}'
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_events_date_type ON events(business_date, event_type);
|
|
CREATE INDEX IF NOT EXISTS idx_events_user_time ON events(platform, platform_user_id, occurred_at_utc);
|
|
CREATE INDEX IF NOT EXISTS idx_signin_date ON signin_events(business_date);
|
|
CREATE INDEX IF NOT EXISTS idx_points_date ON point_transactions(business_date);
|
|
CREATE INDEX IF NOT EXISTS idx_song_date_status ON song_requests(business_date, status);
|
|
CREATE INDEX IF NOT EXISTS idx_playback_date_status ON playback_sessions(business_date, status);
|
|
CREATE INDEX IF NOT EXISTS idx_gifts_date ON gift_events(business_date);
|
|
CREATE INDEX IF NOT EXISTS idx_danmu_date ON danmu_events(business_date);
|
|
CREATE INDEX IF NOT EXISTS idx_tts_date_status ON tts_requests(business_date, status);
|
|
CREATE INDEX IF NOT EXISTS idx_queue_date_action ON queue_events(business_date, action);
|
|
CREATE INDEX IF NOT EXISTS idx_group_date_status ON group_runs(business_date, status);
|
|
CREATE INDEX IF NOT EXISTS idx_bilibili_date_status ON bilibili_connections(business_date, status);
|
|
CREATE INDEX IF NOT EXISTS idx_broadcast_date_status ON broadcast_requests(business_date, status);
|
|
CREATE INDEX IF NOT EXISTS idx_service_date_state ON service_state_events(business_date, new_state);
|
|
CREATE INDEX IF NOT EXISTS idx_admin_audit_date_action ON admin_audit_events(business_date, action);
|
|
CREATE TABLE IF NOT EXISTS live_sessions (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
session_id TEXT NOT NULL UNIQUE,
|
|
kind TEXT NOT NULL,
|
|
started_at_utc TEXT NOT NULL,
|
|
ended_at_utc TEXT,
|
|
status TEXT,
|
|
duration_ms INTEGER,
|
|
source TEXT,
|
|
payload_json TEXT NOT NULL DEFAULT '{}',
|
|
updated_at_utc TEXT NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS live_session_daily_durations (
|
|
session_id TEXT NOT NULL REFERENCES live_sessions(session_id) ON DELETE CASCADE,
|
|
business_date TEXT NOT NULL,
|
|
duration_ms INTEGER NOT NULL,
|
|
PRIMARY KEY(session_id, business_date)
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_live_sessions_kind_time ON live_sessions(kind, started_at_utc);
|
|
CREATE INDEX IF NOT EXISTS idx_live_daily_date ON live_session_daily_durations(business_date);
|
|
"""
|
|
|
|
_DAILY_STATISTICS_VIEW = """
|
|
CREATE VIEW daily_live_statistics AS
|
|
WITH dates AS (
|
|
SELECT business_date FROM live_session_daily_durations
|
|
UNION SELECT business_date FROM events
|
|
UNION SELECT business_date FROM process_sessions
|
|
UNION SELECT business_date FROM signin_events
|
|
UNION SELECT business_date FROM point_transactions
|
|
UNION SELECT business_date FROM queue_events
|
|
UNION SELECT business_date FROM group_runs
|
|
UNION SELECT business_date FROM song_requests
|
|
UNION SELECT business_date FROM playback_sessions
|
|
UNION SELECT business_date FROM gift_events
|
|
UNION SELECT business_date FROM danmu_events
|
|
UNION SELECT business_date FROM tts_requests
|
|
UNION SELECT business_date FROM broadcast_requests
|
|
UNION SELECT business_date FROM bilibili_connections
|
|
UNION SELECT business_date FROM service_state_events
|
|
UNION SELECT business_date FROM admin_audit_events
|
|
)
|
|
SELECT d.business_date,
|
|
COALESCE((SELECT SUM(x.duration_ms) FROM live_session_daily_durations x JOIN live_sessions s ON s.session_id=x.session_id WHERE x.business_date=d.business_date AND s.kind IN ('live','stream','streaming')),0) AS live_duration_ms,
|
|
COALESCE((SELECT SUM(x.duration_ms) FROM live_session_daily_durations x JOIN live_sessions s ON s.session_id=x.session_id WHERE x.business_date=d.business_date AND s.kind='process'),0) AS process_duration_ms,
|
|
COALESCE((SELECT SUM(x.duration_ms) FROM live_session_daily_durations x JOIN live_sessions s ON s.session_id=x.session_id WHERE x.business_date=d.business_date AND s.kind='bilibili_connection'),0) AS bilibili_connection_duration_ms,
|
|
COALESCE((SELECT SUM(reconnect_count) FROM bilibili_connections x WHERE x.business_date=d.business_date),0) AS bilibili_reconnect_count,
|
|
(SELECT COUNT(*) FROM danmu_events x WHERE x.business_date=d.business_date AND x.message_type='danmu') AS danmu_count,
|
|
(SELECT COUNT(*) FROM danmu_events x WHERE x.business_date=d.business_date AND (x.command IS NOT NULL AND x.command<>'') ) AS command_count,
|
|
(SELECT COUNT(DISTINCT platform || ':' || COALESCE(platform_user_id,'')) FROM danmu_events x WHERE x.business_date=d.business_date AND COALESCE(platform_user_id,'')<>'') AS active_users,
|
|
(SELECT COUNT(*) FROM signin_events x WHERE x.business_date=d.business_date) AS signin_count,
|
|
(SELECT COUNT(DISTINCT platform || ':' || COALESCE(platform_user_id,'')) FROM signin_events x WHERE x.business_date=d.business_date AND COALESCE(platform_user_id,'')<>'') AS unique_signin_users,
|
|
COALESCE((SELECT SUM(amount) FROM point_transactions x WHERE x.business_date=d.business_date),0) AS point_net_change,
|
|
(SELECT COUNT(*) FROM queue_events x WHERE x.business_date=d.business_date AND x.action IN ('join','enqueue','entered')) AS queue_join_count,
|
|
(SELECT COUNT(DISTINCT COALESCE(CAST(user_id AS TEXT), json_extract(payload_json,'$.platform_user_id'), item_id)) FROM queue_events x WHERE x.business_date=d.business_date AND x.action IN ('join','enqueue','entered')) AS unique_queue_users,
|
|
(SELECT COUNT(*) FROM queue_events x WHERE x.business_date=d.business_date AND x.action IN ('leave','dequeue','removed')) AS leave_count,
|
|
(SELECT COUNT(*) FROM queue_events x WHERE x.business_date=d.business_date AND x.action IN ('timeout','expired')) AS timeout_count,
|
|
COALESCE((SELECT MAX(queue_size) FROM queue_events x WHERE x.business_date=d.business_date),0) AS max_queue_size,
|
|
(SELECT AVG(wait_ms) FROM queue_events x WHERE x.business_date=d.business_date AND x.action='promoted' AND wait_ms IS NOT NULL) AS avg_wait_ms,
|
|
(SELECT COUNT(*) FROM queue_events x WHERE x.business_date=d.business_date AND x.action='promoted' AND wait_ms IS NOT NULL) AS wait_sample_count,
|
|
(SELECT COUNT(*) FROM group_runs x WHERE x.business_date=d.business_date) AS group_run_count,
|
|
(SELECT COUNT(*) FROM group_runs x WHERE x.business_date=d.business_date AND x.status IN ('success','completed','finished')) AS group_run_success_count,
|
|
(SELECT COUNT(*) FROM group_runs x WHERE x.business_date=d.business_date AND x.status IN ('failed','error')) AS group_run_failed_count,
|
|
COALESCE((SELECT SUM(CASE WHEN ended_at_utc IS NOT NULL THEN MAX(0, CAST((julianday(ended_at_utc)-julianday(started_at_utc))*86400000 AS INTEGER)) ELSE 0 END) FROM group_runs x WHERE x.business_date=d.business_date),0) AS group_run_total_duration_ms,
|
|
(SELECT COUNT(*) FROM song_requests x WHERE x.business_date=d.business_date) AS song_request_count,
|
|
(SELECT COUNT(DISTINCT platform || ':' || COALESCE(platform_user_id,'')) FROM song_requests x WHERE x.business_date=d.business_date AND COALESCE(platform_user_id,'')<>'') AS unique_song_users,
|
|
(SELECT COUNT(*) FROM song_requests x WHERE x.business_date=d.business_date AND x.status IN ('played','completed','finished','success')) AS played_count,
|
|
(SELECT COUNT(*) FROM song_requests x WHERE x.business_date=d.business_date AND x.status IN ('skipped','skip')) AS skipped_count,
|
|
(SELECT COUNT(*) FROM song_requests x WHERE x.business_date=d.business_date AND x.status IN ('error','failed')) AS error_count,
|
|
(SELECT COUNT(*) FROM playback_sessions x WHERE x.business_date=d.business_date AND json_extract(x.payload_json,'$.source')='background') AS background_playback_count,
|
|
(SELECT COUNT(*) FROM gift_events x WHERE x.business_date=d.business_date) AS gift_event_count,
|
|
COALESCE((SELECT SUM(quantity) FROM gift_events x WHERE x.business_date=d.business_date),0) AS gift_quantity,
|
|
(SELECT COUNT(DISTINCT platform || ':' || COALESCE(platform_user_id,'')) FROM gift_events x WHERE x.business_date=d.business_date AND COALESCE(platform_user_id,'')<>'') AS unique_gifters,
|
|
COALESCE((SELECT SUM(total_value) FROM gift_events x WHERE x.business_date=d.business_date AND x.currency='CNY'),0) AS gift_total_value,
|
|
(SELECT COUNT(*) FROM tts_requests x WHERE x.business_date=d.business_date) AS tts_request_count,
|
|
(SELECT COUNT(*) FROM tts_requests x WHERE x.business_date=d.business_date AND x.status IN ('success','completed','finished')) AS tts_success_count,
|
|
(SELECT COUNT(*) FROM tts_requests x WHERE x.business_date=d.business_date AND x.status IN ('failure','failed','error')) AS tts_failure_count,
|
|
COALESCE((SELECT AVG(duration_ms) FROM tts_requests x WHERE x.business_date=d.business_date AND duration_ms IS NOT NULL),0) AS tts_avg_duration_ms,
|
|
(SELECT COUNT(*) FROM broadcast_requests x WHERE x.business_date=d.business_date) AS broadcast_request_count,
|
|
(SELECT COUNT(*) FROM broadcast_requests x WHERE x.business_date=d.business_date AND x.status IN ('success','completed','finished')) AS broadcast_success_count,
|
|
(SELECT COUNT(*) FROM broadcast_requests x WHERE x.business_date=d.business_date AND x.status IN ('failure','failed','error')) AS broadcast_failure_count,
|
|
(SELECT COUNT(*) FROM service_state_events x WHERE x.business_date=d.business_date AND x.new_state IN ('error','failed','failure')) AS service_error_count,
|
|
(SELECT COUNT(*) FROM service_state_events x WHERE x.business_date=d.business_date AND x.new_state IN ('restart','restarted','starting') ) AS service_restart_count,
|
|
(SELECT COUNT(*) FROM admin_audit_events x WHERE x.business_date=d.business_date) AS admin_action_count,
|
|
(SELECT COUNT(*) FROM events x WHERE x.business_date=d.business_date) AS generic_event_count
|
|
FROM dates d;
|
|
"""
|
|
|
|
|
|
class StatsStore:
|
|
"""基于 SQLite 的异步单写统计存储。"""
|
|
|
|
def __init__(
|
|
self,
|
|
database_path: str | Path = DEFAULT_DATABASE_PATH,
|
|
*,
|
|
queue_size: int = 10_000,
|
|
batch_size: int = 100,
|
|
commit_interval: float = 0.5,
|
|
) -> None:
|
|
self.database_path = Path(database_path)
|
|
self.queue: asyncio.Queue[Any] = asyncio.Queue(maxsize=max(1, queue_size))
|
|
self.batch_size = max(1, batch_size)
|
|
self.commit_interval = max(0.05, commit_interval)
|
|
self._writer_task: asyncio.Task[None] | None = None
|
|
self._started = False
|
|
self._closing = False
|
|
self._connection: sqlite3.Connection | None = None
|
|
self._last_error: str | None = None
|
|
self._written = 0
|
|
self._dropped = 0
|
|
self._process_session_id: str | None = None
|
|
|
|
async def start(self) -> bool:
|
|
if self._started and self._writer_task and not self._writer_task.done():
|
|
return True
|
|
if self._closing:
|
|
return False
|
|
try:
|
|
self.database_path.parent.mkdir(parents=True, exist_ok=True)
|
|
connection = sqlite3.connect(self.database_path, timeout=5.0)
|
|
connection.row_factory = sqlite3.Row
|
|
connection.execute("PRAGMA journal_mode=WAL")
|
|
connection.execute("PRAGMA synchronous=NORMAL")
|
|
connection.execute("PRAGMA foreign_keys=ON")
|
|
connection.execute("PRAGMA busy_timeout=5000")
|
|
self._migrate(connection)
|
|
gift_values_repaired = self._repair_gift_value_history(connection)
|
|
if gift_values_repaired:
|
|
LOGGER.warning("已修复统计库历史礼物金额: %s 条", gift_values_repaired)
|
|
repaired = self._repair_startup_reconciliation_history(connection)
|
|
if any(repaired.values()):
|
|
LOGGER.warning("已修复统计库历史异常生命周期: %s", repaired)
|
|
recovered = self._reconcile_stale_lifecycles(connection)
|
|
if any(recovered.values()):
|
|
LOGGER.warning("已收口统计库遗留生命周期: %s", recovered)
|
|
self._connection = connection
|
|
self._started = True
|
|
self._writer_task = asyncio.create_task(self._writer(), name="stats-store-writer")
|
|
return True
|
|
except Exception as exc:
|
|
self._last_error = f"{type(exc).__name__}: {exc}"
|
|
LOGGER.exception("无法启动统计存储")
|
|
try:
|
|
connection.close()
|
|
except (UnboundLocalError, sqlite3.Error):
|
|
pass
|
|
self._connection = None
|
|
self._started = False
|
|
return False
|
|
|
|
async def close(self) -> None:
|
|
if not self._started:
|
|
return
|
|
self._closing = True
|
|
try:
|
|
await self.flush()
|
|
await self.queue.put(_STOP)
|
|
if self._writer_task:
|
|
await self._writer_task
|
|
except Exception as exc:
|
|
self._last_error = f"{type(exc).__name__}: {exc}"
|
|
LOGGER.exception("关闭统计存储失败")
|
|
finally:
|
|
self._started = False
|
|
self._writer_task = None
|
|
if self._connection:
|
|
try:
|
|
self._connection.close()
|
|
except sqlite3.Error:
|
|
pass
|
|
self._connection = None
|
|
self._closing = False
|
|
|
|
async def flush(self) -> bool:
|
|
writer = self._writer_task
|
|
if not self._started or not writer or writer.done():
|
|
return False
|
|
loop = asyncio.get_running_loop()
|
|
completed = loop.create_future()
|
|
try:
|
|
await self.queue.put((_FLUSH, completed))
|
|
done, _ = await asyncio.wait((completed, writer), return_when=asyncio.FIRST_COMPLETED)
|
|
if completed in done:
|
|
return bool(completed.result())
|
|
self._last_error = "统计 writer 在 flush 完成前退出"
|
|
return False
|
|
except Exception as exc:
|
|
self._last_error = f"{type(exc).__name__}: {exc}"
|
|
return False
|
|
|
|
async def execute(self, func, *args) -> Any:
|
|
"""在 writer 持有的唯一连接上执行同步函数并返回其结果。
|
|
|
|
供其它模块（如兑换码）复用同一 sqlite 连接、串行化写入，
|
|
消除多连接写同一文件导致的 "database is locked" 写争用。
|
|
"""
|
|
writer = self._writer_task
|
|
if not self._started or not writer or writer.done():
|
|
raise RuntimeError("统计存储未启动")
|
|
loop = asyncio.get_running_loop()
|
|
completed = loop.create_future()
|
|
try:
|
|
await self.queue.put(("__execute__", func, args, completed))
|
|
except Exception as exc:
|
|
completed.cancel()
|
|
raise RuntimeError("统计存储写入队列已满") from exc
|
|
return await completed
|
|
|
|
def emit_nowait(
|
|
self,
|
|
event_type: str,
|
|
payload: Mapping[str, Any] | None = None,
|
|
**fields: Any,
|
|
) -> bool:
|
|
"""无等待旁路写入；队列满或未启动时返回 False。"""
|
|
if not self._started or self._closing or not event_type:
|
|
self._dropped += 1
|
|
return False
|
|
try:
|
|
operation = self._event_operation(event_type, payload, fields)
|
|
self.queue.put_nowait(operation)
|
|
return True
|
|
except (asyncio.QueueFull, Exception) as exc:
|
|
self._dropped += 1
|
|
self._last_error = f"{type(exc).__name__}: {exc}"
|
|
return False
|
|
|
|
async def emit(
|
|
self,
|
|
event_type: str,
|
|
payload: Mapping[str, Any] | None = None,
|
|
**fields: Any,
|
|
) -> bool:
|
|
"""等待队列容量后旁路写入；失败不抛出。"""
|
|
if not self._started or self._closing or not event_type:
|
|
self._dropped += 1
|
|
return False
|
|
try:
|
|
await self.queue.put(self._event_operation(event_type, payload, fields))
|
|
return True
|
|
except Exception as exc:
|
|
self._dropped += 1
|
|
self._last_error = f"{type(exc).__name__}: {exc}"
|
|
return False
|
|
|
|
def _event_operation(
|
|
self,
|
|
event_type: str,
|
|
payload: Mapping[str, Any] | None,
|
|
fields: Mapping[str, Any],
|
|
) -> tuple[str, dict[str, Any]]:
|
|
now = _utc_iso(fields.get("occurred_at_utc") or utc_now())
|
|
event_id = str(fields.get("event_id") or self.make_event_id(event_type, now, fields, payload or {}))
|
|
return (
|
|
"events",
|
|
{
|
|
"event_id": event_id,
|
|
"event_type": event_type,
|
|
"category": fields.get("category"),
|
|
"occurred_at_utc": now,
|
|
"business_date": fields.get("business_date") or business_date(now),
|
|
"process_session_id": fields.get("process_session_id") or self._process_session_id,
|
|
"user_id": fields.get("user_id"),
|
|
"platform": fields.get("platform"),
|
|
"platform_user_id": fields.get("platform_user_id"),
|
|
"correlation_id": fields.get("correlation_id"),
|
|
"payload_json": _json(payload or {}),
|
|
},
|
|
)
|
|
|
|
def _enqueue_record(self, table: str, fields: Mapping[str, Any]) -> bool:
|
|
if not self._started or self._closing:
|
|
self._dropped += 1
|
|
return False
|
|
try:
|
|
self.queue.put_nowait((table, _clean_fields(fields)))
|
|
return True
|
|
except Exception as exc:
|
|
self._dropped += 1
|
|
self._last_error = f"{type(exc).__name__}: {exc}"
|
|
return False
|
|
|
|
async def _writer(self) -> None:
|
|
connection = self._connection
|
|
if connection is None:
|
|
return
|
|
pending = 0
|
|
while True:
|
|
try:
|
|
try:
|
|
item = await asyncio.wait_for(self.queue.get(), timeout=self.commit_interval)
|
|
except TimeoutError:
|
|
if pending:
|
|
connection.commit()
|
|
pending = 0
|
|
continue
|
|
if item is _STOP:
|
|
if pending:
|
|
connection.commit()
|
|
self.queue.task_done()
|
|
return
|
|
if isinstance(item, tuple) and item and item[0] is _FLUSH:
|
|
if pending:
|
|
connection.commit()
|
|
pending = 0
|
|
future = item[1]
|
|
if not future.done():
|
|
future.set_result(True)
|
|
self.queue.task_done()
|
|
continue
|
|
if isinstance(item, tuple) and len(item) == 4 and item[0] == "__execute__":
|
|
if pending:
|
|
connection.commit()
|
|
pending = 0
|
|
_, func, args, future = item
|
|
try:
|
|
result = func(connection, *args)
|
|
if not future.done():
|
|
future.set_result(result)
|
|
except Exception as exc:
|
|
try:
|
|
connection.rollback()
|
|
except Exception:
|
|
pass
|
|
if not future.done():
|
|
future.set_exception(exc)
|
|
self.queue.task_done()
|
|
continue
|
|
table, fields = item
|
|
if table == "__process_start__":
|
|
self._insert(connection, "process_sessions", fields)
|
|
self._upsert_live_session(connection, {
|
|
"session_id": fields["session_id"], "kind": "process",
|
|
"started_at_utc": fields["started_at_utc"], "status": "running",
|
|
"source": "process_sessions", "payload_json": fields.get("metadata_json", "{}"),
|
|
})
|
|
elif table == "__process_end__":
|
|
connection.execute(
|
|
"UPDATE process_sessions SET ended_at_utc=?, exit_reason=? WHERE session_id=?",
|
|
(fields["ended_at_utc"], fields.get("exit_reason"), fields["session_id"]),
|
|
)
|
|
row = connection.execute(
|
|
"SELECT started_at_utc, metadata_json FROM process_sessions WHERE session_id=?",
|
|
(fields["session_id"],),
|
|
).fetchone()
|
|
if row:
|
|
self._upsert_live_session(connection, {
|
|
"session_id": fields["session_id"], "kind": "process",
|
|
"started_at_utc": row[0], "ended_at_utc": fields["ended_at_utc"],
|
|
"status": "ended", "source": "process_sessions", "payload_json": row[1],
|
|
})
|
|
elif table == "__bilibili_connection__":
|
|
self._upsert_lifecycle(connection, "bilibili_connections", fields)
|
|
if fields.get("disconnected_at_utc"):
|
|
row = connection.execute(
|
|
"SELECT connected_at_utc, disconnected_at_utc, status, payload_json "
|
|
"FROM bilibili_connections WHERE connection_id=?",
|
|
(fields["connection_id"],),
|
|
).fetchone()
|
|
if row and row[1]:
|
|
self._upsert_live_session(connection, {
|
|
"session_id": fields["connection_id"], "kind": "bilibili_connection",
|
|
"started_at_utc": row[0], "ended_at_utc": row[1],
|
|
"status": row[2], "source": "bilibili_connections",
|
|
"payload_json": row[3],
|
|
})
|
|
elif table == "__user_upsert__":
|
|
self._upsert_user(connection, fields)
|
|
elif table == "__import__":
|
|
self._run_import(connection, fields)
|
|
pending = 0
|
|
elif table == "__live_session_upsert__":
|
|
self._upsert_live_session(connection, fields)
|
|
else:
|
|
if table in _LIFECYCLE_KEYS:
|
|
inserted = self._upsert_lifecycle(connection, table, fields)
|
|
else:
|
|
inserted = self._insert(connection, table, fields)
|
|
if table == "gift_events" and inserted:
|
|
self._aggregate_gift(connection, fields)
|
|
pending += 1
|
|
self._written += 1
|
|
self.queue.task_done()
|
|
if pending >= self.batch_size:
|
|
connection.commit()
|
|
pending = 0
|
|
except Exception as exc:
|
|
connection.rollback()
|
|
pending = 0
|
|
self._last_error = f"{type(exc).__name__}: {exc}"
|
|
LOGGER.exception("统计写入失败")
|
|
try:
|
|
self.queue.task_done()
|
|
except ValueError:
|
|
pass
|
|
|
|
@staticmethod
|
|
def _insert(connection: sqlite3.Connection, table: str, fields: Mapping[str, Any]) -> bool:
|
|
allowed = {
|
|
"process_sessions", "signin_events", "point_transactions", "queue_events",
|
|
"login_sessions", "group_runs", "song_requests", "playback_sessions",
|
|
"gift_events", "broadcast_requests", "tts_requests", "bilibili_connections",
|
|
"danmu_events", "service_state_events", "admin_audit_events", "events",
|
|
}
|
|
if table not in allowed:
|
|
raise ValueError(f"不允许写入表: {table}")
|
|
columns = tuple(fields)
|
|
sql = f"INSERT OR IGNORE INTO {table} ({','.join(columns)}) VALUES ({','.join('?' for _ in columns)})"
|
|
cursor = connection.execute(sql, tuple(fields[column] for column in columns))
|
|
return cursor.rowcount > 0
|
|
|
|
@staticmethod
|
|
def _upsert_lifecycle(connection: sqlite3.Connection, table: str, fields: Mapping[str, Any]) -> bool:
|
|
definition = _LIFECYCLE_KEYS.get(table)
|
|
if definition is None:
|
|
raise ValueError(f"不允许生命周期更新表: {table}")
|
|
key_column, start_column, business_column = definition
|
|
if key_column not in fields:
|
|
raise ValueError(f"{table} 缺少稳定 ID: {key_column}")
|
|
columns = tuple(fields)
|
|
placeholders = ",".join("?" for _ in columns)
|
|
immutable = {key_column, start_column, business_column}
|
|
updates = tuple(column for column in columns if column not in immutable)
|
|
if updates:
|
|
update_sql = ",".join(f"{column}=excluded.{column}" for column in updates)
|
|
conflict_sql = f"DO UPDATE SET {update_sql}"
|
|
else:
|
|
conflict_sql = "DO NOTHING"
|
|
sql = (
|
|
f"INSERT INTO {table} ({','.join(columns)}) VALUES ({placeholders}) "
|
|
f"ON CONFLICT({key_column}) {conflict_sql}"
|
|
)
|
|
cursor = connection.execute(sql, tuple(fields[column] for column in columns))
|
|
return cursor.rowcount > 0
|
|
|
|
@staticmethod
|
|
def _upsert_user(connection: sqlite3.Connection, fields: Mapping[str, Any]) -> None:
|
|
connection.execute(
|
|
"""INSERT INTO users (
|
|
platform, platform_user_id, display_name, avatar_url, user_level, is_admin,
|
|
first_seen_at_utc, last_seen_at_utc, snapshot_json
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(platform, platform_user_id) DO UPDATE SET
|
|
display_name=CASE WHEN excluded.last_seen_at_utc >= users.last_seen_at_utc
|
|
THEN COALESCE(excluded.display_name, users.display_name) ELSE users.display_name END,
|
|
avatar_url=CASE WHEN excluded.last_seen_at_utc >= users.last_seen_at_utc
|
|
THEN COALESCE(excluded.avatar_url, users.avatar_url) ELSE users.avatar_url END,
|
|
user_level=CASE WHEN excluded.last_seen_at_utc >= users.last_seen_at_utc
|
|
THEN COALESCE(excluded.user_level, users.user_level) ELSE users.user_level END,
|
|
is_admin=CASE WHEN excluded.last_seen_at_utc >= users.last_seen_at_utc
|
|
THEN excluded.is_admin ELSE users.is_admin END,
|
|
first_seen_at_utc=MIN(users.first_seen_at_utc, excluded.first_seen_at_utc),
|
|
last_seen_at_utc=MAX(users.last_seen_at_utc, excluded.last_seen_at_utc),
|
|
snapshot_json=CASE WHEN excluded.last_seen_at_utc >= users.last_seen_at_utc
|
|
THEN excluded.snapshot_json ELSE users.snapshot_json END""",
|
|
tuple(fields[name] for name in (
|
|
"platform", "platform_user_id", "display_name", "avatar_url", "user_level",
|
|
"is_admin", "first_seen_at_utc", "last_seen_at_utc", "snapshot_json",
|
|
)),
|
|
)
|
|
|
|
@staticmethod
|
|
def _upsert_live_session(connection: sqlite3.Connection, fields: Mapping[str, Any]) -> None:
|
|
existing = connection.execute(
|
|
"SELECT kind, started_at_utc, ended_at_utc, status, duration_ms, source, payload_json FROM live_sessions WHERE session_id=?",
|
|
(fields["session_id"],),
|
|
).fetchone()
|
|
started = fields.get("started_at_utc") or (existing[1] if existing else None)
|
|
if not started:
|
|
raise ValueError("live session 缺少 started_at_utc")
|
|
ended = fields.get("ended_at_utc") if "ended_at_utc" in fields else (existing[2] if existing else None)
|
|
kind = fields.get("kind") or (existing[0] if existing else "live")
|
|
status = fields.get("status") if "status" in fields else (existing[3] if existing else None)
|
|
source = fields.get("source") if "source" in fields else (existing[5] if existing else None)
|
|
payload_json = fields.get("payload_json") if "payload_json" in fields else (existing[6] if existing else "{}")
|
|
duration_ms = fields.get("duration_ms")
|
|
if duration_ms is None and ended:
|
|
duration_ms = StatsStore._duration_ms(started, ended)
|
|
connection.execute(
|
|
"""INSERT INTO live_sessions (
|
|
session_id, kind, started_at_utc, ended_at_utc, status, duration_ms,
|
|
source, payload_json, updated_at_utc
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(session_id) DO UPDATE SET
|
|
kind=excluded.kind, started_at_utc=excluded.started_at_utc,
|
|
ended_at_utc=excluded.ended_at_utc, status=excluded.status,
|
|
duration_ms=excluded.duration_ms, source=excluded.source,
|
|
payload_json=excluded.payload_json, updated_at_utc=excluded.updated_at_utc""",
|
|
(fields["session_id"], kind, started, ended, status, duration_ms, source, payload_json, utc_now()),
|
|
)
|
|
connection.execute("DELETE FROM live_session_daily_durations WHERE session_id=?", (fields["session_id"],))
|
|
if ended:
|
|
for day, allocated_ms in StatsStore._split_session_duration(started, ended, duration_ms):
|
|
connection.execute(
|
|
"INSERT INTO live_session_daily_durations(session_id, business_date, duration_ms) VALUES (?, ?, ?)",
|
|
(fields["session_id"], day, allocated_ms),
|
|
)
|
|
|
|
@staticmethod
|
|
def _parse_utc(value: str) -> datetime:
|
|
parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
|
|
if parsed.tzinfo is None:
|
|
parsed = parsed.replace(tzinfo=UTC)
|
|
return parsed.astimezone(UTC)
|
|
|
|
@staticmethod
|
|
def _duration_ms(started_at_utc: str, ended_at_utc: str) -> int:
|
|
return max(0, round((StatsStore._parse_utc(ended_at_utc) - StatsStore._parse_utc(started_at_utc)).total_seconds() * 1000))
|
|
|
|
@staticmethod
|
|
def _split_session_duration(started_at_utc: str, ended_at_utc: str, duration_ms: Any = None) -> list[tuple[str, int]]:
|
|
started = StatsStore._parse_utc(started_at_utc)
|
|
ended = StatsStore._parse_utc(ended_at_utc)
|
|
if ended <= started:
|
|
return []
|
|
exact_total = (ended - started).total_seconds() * 1000
|
|
target_total = max(0, int(duration_ms)) if duration_ms is not None else round(exact_total)
|
|
segments: list[tuple[str, float]] = []
|
|
cursor = started
|
|
while cursor < ended:
|
|
local = cursor.astimezone(BEIJING_TZ)
|
|
boundary_date = local.date() if local.hour < 4 else local.date() + timedelta(days=1)
|
|
boundary = datetime.combine(boundary_date, datetime.min.time(), BEIJING_TZ).replace(hour=4).astimezone(UTC)
|
|
segment_end = min(ended, boundary)
|
|
segments.append((business_date(cursor), (segment_end - cursor).total_seconds() * 1000))
|
|
cursor = segment_end
|
|
allocated: list[tuple[str, int]] = []
|
|
consumed = 0
|
|
for index, (day, milliseconds) in enumerate(segments):
|
|
value = target_total - consumed if index == len(segments) - 1 else round(target_total * milliseconds / exact_total)
|
|
value = max(0, value)
|
|
allocated.append((day, value))
|
|
consumed += value
|
|
return allocated
|
|
|
|
@staticmethod
|
|
def _business_day_end(value: datetime) -> datetime:
|
|
local = value.astimezone(BEIJING_TZ)
|
|
boundary_date = local.date() if local.hour < 4 else local.date() + timedelta(days=1)
|
|
return datetime.combine(boundary_date, datetime.min.time(), BEIJING_TZ).replace(hour=4).astimezone(UTC)
|
|
|
|
@classmethod
|
|
def _merged_daily_interval_durations(
|
|
cls,
|
|
intervals: Sequence[tuple[Any, Any]],
|
|
) -> dict[str, int]:
|
|
daily_intervals: dict[str, list[tuple[datetime, datetime]]] = {}
|
|
for started_value, ended_value in intervals:
|
|
if not started_value or not ended_value:
|
|
continue
|
|
try:
|
|
started = cls._parse_utc(str(started_value))
|
|
ended = cls._parse_utc(str(ended_value))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if ended <= started:
|
|
continue
|
|
cursor = started
|
|
while cursor < ended:
|
|
segment_end = min(ended, cls._business_day_end(cursor))
|
|
daily_intervals.setdefault(business_date(cursor), []).append((cursor, segment_end))
|
|
cursor = segment_end
|
|
|
|
durations: dict[str, int] = {}
|
|
for day, segments in daily_intervals.items():
|
|
merged_ms = 0.0
|
|
current_start: datetime | None = None
|
|
current_end: datetime | None = None
|
|
for segment_start, segment_end in sorted(segments):
|
|
if current_start is None:
|
|
current_start, current_end = segment_start, segment_end
|
|
elif segment_start <= current_end:
|
|
current_end = max(current_end, segment_end)
|
|
else:
|
|
merged_ms += (current_end - current_start).total_seconds() * 1000
|
|
current_start, current_end = segment_start, segment_end
|
|
if current_start is not None and current_end is not None:
|
|
merged_ms += (current_end - current_start).total_seconds() * 1000
|
|
durations[day] = min(86_400_000, max(0, round(merged_ms)))
|
|
return durations
|
|
|
|
@staticmethod
|
|
def _aggregate_gift(connection: sqlite3.Connection, fields: Mapping[str, Any]) -> None:
|
|
connection.execute(
|
|
"""INSERT INTO gift_aggregates (
|
|
business_date, platform, platform_user_id, gift_id, gift_name,
|
|
quantity, total_value, updated_at_utc
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(business_date, platform, platform_user_id, gift_id) DO UPDATE SET
|
|
gift_name=COALESCE(excluded.gift_name, gift_aggregates.gift_name),
|
|
quantity=gift_aggregates.quantity + excluded.quantity,
|
|
total_value=gift_aggregates.total_value + excluded.total_value,
|
|
updated_at_utc=excluded.updated_at_utc""",
|
|
(
|
|
fields["business_date"], fields.get("platform") or "",
|
|
fields.get("platform_user_id") or "", fields.get("gift_id") or "",
|
|
fields.get("gift_name"), fields.get("quantity", 1),
|
|
fields.get("total_value", 0), fields.get("occurred_at_utc") or utc_now(),
|
|
),
|
|
)
|
|
|
|
def _run_import(self, connection: sqlite3.Connection, fields: Mapping[str, Any]) -> None:
|
|
future = fields["future"]
|
|
try:
|
|
existing = connection.execute(
|
|
"SELECT id, source_fingerprint, status FROM import_sources WHERE source_key=?",
|
|
(fields["source_key"],),
|
|
).fetchone()
|
|
if existing and existing[1] == fields.get("fingerprint") and existing[2] == "completed":
|
|
connection.commit()
|
|
if not future.done():
|
|
future.set_result(True)
|
|
return
|
|
now = utc_now()
|
|
connection.execute(
|
|
"""INSERT INTO import_sources (
|
|
source_key, source_type, source_fingerprint, first_imported_at_utc,
|
|
last_imported_at_utc, status, metadata_json
|
|
) VALUES (?, ?, ?, ?, ?, 'running', ?)
|
|
ON CONFLICT(source_key) DO UPDATE SET
|
|
source_type=excluded.source_type,
|
|
source_fingerprint=excluded.source_fingerprint,
|
|
last_imported_at_utc=excluded.last_imported_at_utc,
|
|
status='running', metadata_json=excluded.metadata_json""",
|
|
(fields["source_key"], fields["source_type"], fields.get("fingerprint"), now, now, fields["metadata_json"]),
|
|
)
|
|
source_id = connection.execute(
|
|
"SELECT id FROM import_sources WHERE source_key=?", (fields["source_key"],)
|
|
).fetchone()[0]
|
|
for table, record in fields["records"]:
|
|
if table == "users":
|
|
self._upsert_user(connection, record)
|
|
else:
|
|
if table in _LIFECYCLE_KEYS:
|
|
inserted = self._upsert_lifecycle(connection, table, record)
|
|
else:
|
|
inserted = self._insert(connection, table, record)
|
|
if table == "gift_events" and inserted:
|
|
self._aggregate_gift(connection, record)
|
|
connection.execute(
|
|
"""INSERT INTO import_checkpoints (
|
|
source_id, checkpoint_key, checkpoint_value, updated_at_utc, payload_json
|
|
) VALUES (?, ?, ?, ?, '{}')
|
|
ON CONFLICT(source_id, checkpoint_key) DO UPDATE SET
|
|
checkpoint_value=excluded.checkpoint_value, updated_at_utc=excluded.updated_at_utc""",
|
|
(source_id, fields["checkpoint_key"], fields.get("checkpoint_value"), now),
|
|
)
|
|
connection.execute(
|
|
"UPDATE import_sources SET status='completed', last_imported_at_utc=? WHERE id=?",
|
|
(now, source_id),
|
|
)
|
|
connection.commit()
|
|
if not future.done():
|
|
future.set_result(True)
|
|
except Exception as exc:
|
|
connection.rollback()
|
|
if not future.done():
|
|
future.set_result(False)
|
|
raise exc
|
|
|
|
@staticmethod
|
|
def _migrate(connection: sqlite3.Connection) -> None:
|
|
current_version = int(connection.execute("PRAGMA user_version").fetchone()[0])
|
|
if current_version > _SCHEMA_VERSION:
|
|
raise RuntimeError(f"数据库版本 {current_version} 高于程序支持版本 {_SCHEMA_VERSION}")
|
|
connection.executescript(_SCHEMA)
|
|
connection.execute("DROP VIEW IF EXISTS daily_live_statistics")
|
|
connection.executescript(_DAILY_STATISTICS_VIEW)
|
|
# 积分权威列迁移：旧库补充 points 列，并从历史 snapshot_json 一次性回填。
|
|
user_columns = {row[1] for row in connection.execute("PRAGMA table_info(users)")}
|
|
if "points" not in user_columns:
|
|
connection.execute("ALTER TABLE users ADD COLUMN points INTEGER NOT NULL DEFAULT 0")
|
|
connection.execute(
|
|
"UPDATE users SET points = CAST(COALESCE(json_extract(snapshot_json, '$.points'), 0) AS INTEGER)"
|
|
)
|
|
for version in range(max(1, current_version + 1), _SCHEMA_VERSION + 1):
|
|
connection.execute(
|
|
"INSERT OR IGNORE INTO schema_migrations(version, applied_at_utc) VALUES (?, ?)",
|
|
(version, utc_now()),
|
|
)
|
|
connection.execute(f"PRAGMA user_version={_SCHEMA_VERSION}")
|
|
connection.commit()
|
|
|
|
@staticmethod
|
|
def _pid_is_running(pid: Any) -> bool:
|
|
try:
|
|
value = int(pid)
|
|
except (TypeError, ValueError):
|
|
return False
|
|
if value <= 0:
|
|
return False
|
|
if os.name == "nt":
|
|
process = ctypes.windll.kernel32.OpenProcess(0x1000, False, value)
|
|
if not process:
|
|
return False
|
|
try:
|
|
exit_code = ctypes.c_ulong()
|
|
if not ctypes.windll.kernel32.GetExitCodeProcess(process, ctypes.byref(exit_code)):
|
|
return False
|
|
return exit_code.value == 259
|
|
finally:
|
|
ctypes.windll.kernel32.CloseHandle(process)
|
|
try:
|
|
os.kill(value, 0)
|
|
return True
|
|
except PermissionError:
|
|
return True
|
|
except (OSError, ProcessLookupError):
|
|
return False
|
|
|
|
@staticmethod
|
|
def _next_lifecycle_start(
|
|
connection: sqlite3.Connection,
|
|
table: str,
|
|
start_column: str,
|
|
row_id: int,
|
|
started_at_utc: str,
|
|
) -> str | None:
|
|
row = connection.execute(
|
|
f"SELECT {start_column} FROM {table} "
|
|
f"WHERE {start_column}>? OR ({start_column}=? AND id>?) "
|
|
f"ORDER BY {start_column}, id LIMIT 1",
|
|
(started_at_utc, started_at_utc, row_id),
|
|
).fetchone()
|
|
return str(row[0]) if row else None
|
|
|
|
@classmethod
|
|
def _reconciliation_end(
|
|
cls,
|
|
connection: sqlite3.Connection,
|
|
table: str,
|
|
start_column: str,
|
|
row_id: int,
|
|
started_at_utc: str,
|
|
now: str,
|
|
) -> str:
|
|
next_started = cls._next_lifecycle_start(
|
|
connection, table, start_column, row_id, started_at_utc
|
|
)
|
|
if next_started and cls._parse_utc(next_started) < cls._parse_utc(now):
|
|
return next_started
|
|
return now
|
|
|
|
@staticmethod
|
|
def _repair_gift_value_history(connection: sqlite3.Connection) -> int:
|
|
cursor = connection.execute(
|
|
"""UPDATE gift_events SET
|
|
total_value=MAX(0, CAST(json_extract(payload_json, '$.raw_total_coin') AS REAL)) / 1000.0,
|
|
unit_value=CASE WHEN quantity > 0 THEN
|
|
(MAX(0, CAST(json_extract(payload_json, '$.raw_total_coin') AS REAL)) / 1000.0) / quantity
|
|
ELSE 0 END,
|
|
payload_json=json_set(
|
|
payload_json, '$.value_rule', 'bilibili_gold_coin_1000_to_cny_1_v2'
|
|
)
|
|
WHERE currency='CNY'
|
|
AND LOWER(COALESCE(json_extract(payload_json, '$.coin_type'), ''))='gold'
|
|
AND COALESCE(json_extract(payload_json, '$.value_rule'), '')<>'bilibili_gold_coin_1000_to_cny_1_v2'
|
|
AND (
|
|
json_extract(payload_json, '$.value_rule')='gold_battery_10_to_cny_1_v1'
|
|
OR json_extract(payload_json, '$.value_migration')='bilibili_coin_to_cny_v1'
|
|
)
|
|
AND json_extract(payload_json, '$.raw_total_coin') IS NOT NULL"""
|
|
)
|
|
repaired = max(0, int(cursor.rowcount))
|
|
if repaired:
|
|
connection.execute("DELETE FROM gift_aggregates")
|
|
connection.execute(
|
|
"""INSERT INTO gift_aggregates (
|
|
business_date, platform, platform_user_id, gift_id, gift_name,
|
|
quantity, total_value, updated_at_utc
|
|
)
|
|
SELECT business_date, COALESCE(platform, ''), COALESCE(platform_user_id, ''),
|
|
COALESCE(gift_id, ''), MAX(gift_name), SUM(quantity), SUM(total_value),
|
|
MAX(occurred_at_utc)
|
|
FROM gift_events
|
|
GROUP BY business_date, COALESCE(platform, ''),
|
|
COALESCE(platform_user_id, ''), COALESCE(gift_id, '')"""
|
|
)
|
|
connection.commit()
|
|
return repaired
|
|
|
|
@classmethod
|
|
def _repair_startup_reconciliation_history(cls, connection: sqlite3.Connection) -> dict[str, int]:
|
|
repaired = {"process_sessions": 0, "bilibili_connections": 0, "group_runs": 0}
|
|
|
|
process_rows = connection.execute(
|
|
"SELECT id, session_id, started_at_utc, ended_at_utc, exit_reason, metadata_json "
|
|
"FROM process_sessions ORDER BY started_at_utc, id"
|
|
).fetchall()
|
|
for index, row in enumerate(process_rows[:-1]):
|
|
next_started = str(process_rows[index + 1][2])
|
|
if row[4] != "startup_reconciliation" or not row[3]:
|
|
continue
|
|
if cls._parse_utc(str(row[3])) <= cls._parse_utc(next_started):
|
|
continue
|
|
connection.execute(
|
|
"UPDATE process_sessions SET ended_at_utc=? WHERE id=?",
|
|
(next_started, row[0]),
|
|
)
|
|
cls._upsert_live_session(connection, {
|
|
"session_id": row[1], "kind": "process", "started_at_utc": row[2],
|
|
"ended_at_utc": next_started, "status": "interrupted",
|
|
"source": "process_sessions", "payload_json": row[5] or "{}",
|
|
})
|
|
repaired["process_sessions"] += 1
|
|
|
|
connection_rows = connection.execute(
|
|
"SELECT id, connection_id, connected_at_utc, disconnected_at_utc, disconnect_reason, payload_json "
|
|
"FROM bilibili_connections ORDER BY connected_at_utc, id"
|
|
).fetchall()
|
|
for index, row in enumerate(connection_rows[:-1]):
|
|
next_started = str(connection_rows[index + 1][2])
|
|
if row[4] != "startup_reconciliation" or not row[3]:
|
|
continue
|
|
if cls._parse_utc(str(row[3])) <= cls._parse_utc(next_started):
|
|
continue
|
|
connection.execute(
|
|
"UPDATE bilibili_connections SET disconnected_at_utc=? WHERE id=?",
|
|
(next_started, row[0]),
|
|
)
|
|
cls._upsert_live_session(connection, {
|
|
"session_id": row[1], "kind": "bilibili_connection", "started_at_utc": row[2],
|
|
"ended_at_utc": next_started, "status": "failed",
|
|
"source": "bilibili_connections", "payload_json": row[5] or "{}",
|
|
})
|
|
repaired["bilibili_connections"] += 1
|
|
|
|
group_rows = connection.execute(
|
|
"SELECT id, started_at_utc, ended_at_utc, payload_json "
|
|
"FROM group_runs ORDER BY started_at_utc, id"
|
|
).fetchall()
|
|
for index, row in enumerate(group_rows[:-1]):
|
|
if not row[2]:
|
|
continue
|
|
try:
|
|
payload = json.loads(row[3] or "{}")
|
|
except (TypeError, ValueError, json.JSONDecodeError):
|
|
continue
|
|
next_started = str(group_rows[index + 1][1])
|
|
if payload.get("reason") != "startup_reconciliation":
|
|
continue
|
|
if cls._parse_utc(str(row[2])) <= cls._parse_utc(next_started):
|
|
continue
|
|
connection.execute(
|
|
"UPDATE group_runs SET ended_at_utc=? WHERE id=?",
|
|
(next_started, row[0]),
|
|
)
|
|
repaired["group_runs"] += 1
|
|
|
|
connection.commit()
|
|
return repaired
|
|
|
|
@classmethod
|
|
def _reconcile_stale_lifecycles(cls, connection: sqlite3.Connection) -> dict[str, int]:
|
|
"""启动时收口异常退出遗留的进程、B站连接和配置组会话。"""
|
|
recovered = {"process_sessions": 0, "bilibili_connections": 0, "group_runs": 0}
|
|
now = utc_now()
|
|
stale_process_ids: list[str] = []
|
|
for row in connection.execute(
|
|
"SELECT id, session_id, pid, started_at_utc, metadata_json "
|
|
"FROM process_sessions WHERE ended_at_utc IS NULL ORDER BY started_at_utc, id"
|
|
).fetchall():
|
|
if cls._pid_is_running(row[2]):
|
|
continue
|
|
ended_at = cls._reconciliation_end(
|
|
connection, "process_sessions", "started_at_utc", row[0], row[3], now
|
|
)
|
|
connection.execute(
|
|
"UPDATE process_sessions SET ended_at_utc=?, exit_reason=? WHERE session_id=? AND ended_at_utc IS NULL",
|
|
(ended_at, "startup_reconciliation", row[1]),
|
|
)
|
|
cls._upsert_live_session(connection, {
|
|
"session_id": row[1], "kind": "process", "started_at_utc": row[3],
|
|
"ended_at_utc": ended_at, "status": "interrupted", "source": "process_sessions",
|
|
"payload_json": row[4] or "{}",
|
|
})
|
|
stale_process_ids.append(str(row[1]))
|
|
recovered["process_sessions"] += 1
|
|
|
|
live_process_exists = any(
|
|
cls._pid_is_running(row[0])
|
|
for row in connection.execute(
|
|
"SELECT pid FROM process_sessions WHERE ended_at_utc IS NULL"
|
|
).fetchall()
|
|
)
|
|
if live_process_exists:
|
|
connection.commit()
|
|
return recovered
|
|
|
|
for row in connection.execute(
|
|
"SELECT id, connection_id, connected_at_utc, payload_json FROM bilibili_connections "
|
|
"WHERE disconnected_at_utc IS NULL ORDER BY connected_at_utc, id"
|
|
).fetchall():
|
|
ended_at = cls._reconciliation_end(
|
|
connection, "bilibili_connections", "connected_at_utc", row[0], row[2], now
|
|
)
|
|
connection.execute(
|
|
"UPDATE bilibili_connections SET disconnected_at_utc=?, status='failed', disconnect_reason=? "
|
|
"WHERE connection_id=? AND disconnected_at_utc IS NULL",
|
|
(ended_at, "startup_reconciliation", row[1]),
|
|
)
|
|
cls._upsert_live_session(connection, {
|
|
"session_id": row[1], "kind": "bilibili_connection", "started_at_utc": row[2],
|
|
"ended_at_utc": ended_at, "status": "failed", "source": "bilibili_connections",
|
|
"payload_json": row[3] or "{}",
|
|
})
|
|
recovered["bilibili_connections"] += 1
|
|
|
|
for row in connection.execute(
|
|
"SELECT id, group_run_id, started_at_utc, payload_json FROM group_runs "
|
|
"WHERE ended_at_utc IS NULL AND status='running' ORDER BY started_at_utc, id"
|
|
).fetchall():
|
|
try:
|
|
payload = json.loads(row[3] or "{}")
|
|
except (TypeError, ValueError, json.JSONDecodeError):
|
|
payload = {}
|
|
payload["reason"] = "startup_reconciliation"
|
|
if stale_process_ids:
|
|
payload["stale_process_session_ids"] = stale_process_ids
|
|
ended_at = cls._reconciliation_end(
|
|
connection, "group_runs", "started_at_utc", row[0], row[2], now
|
|
)
|
|
connection.execute(
|
|
"UPDATE group_runs SET ended_at_utc=?, status='interrupted', payload_json=? "
|
|
"WHERE group_run_id=? AND ended_at_utc IS NULL",
|
|
(ended_at, _json(payload), row[1]),
|
|
)
|
|
recovered["group_runs"] += 1
|
|
|
|
connection.commit()
|
|
return recovered
|
|
|
|
@staticmethod
|
|
def make_event_id(*parts: Any) -> str:
|
|
raw = _json(parts).encode("utf-8", "replace")
|
|
return hashlib.sha256(raw).hexdigest()
|
|
|
|
def record_process_start(
|
|
self,
|
|
session_id: str,
|
|
*,
|
|
pid: int | None = None,
|
|
host: str | None = None,
|
|
app_version: str | None = None,
|
|
metadata: Mapping[str, Any] | None = None,
|
|
started_at_utc: str | None = None,
|
|
) -> bool:
|
|
self._process_session_id = session_id
|
|
started = started_at_utc or utc_now()
|
|
return self._enqueue_record("__process_start__", {
|
|
"session_id": session_id, "started_at_utc": started, "business_date": business_date(started),
|
|
"pid": pid, "host": host, "app_version": app_version, "metadata_json": _json(metadata or {}),
|
|
})
|
|
|
|
def record_process_end(self, session_id: str | None = None, *, exit_reason: str | None = None) -> bool:
|
|
target = session_id or self._process_session_id
|
|
if not target or not self._started or self._closing:
|
|
return False
|
|
try:
|
|
self.queue.put_nowait(("__process_end__", {"session_id": target, "ended_at_utc": utc_now(), "exit_reason": exit_reason}))
|
|
if target == self._process_session_id:
|
|
self._process_session_id = None
|
|
return True
|
|
except Exception:
|
|
self._dropped += 1
|
|
return False
|
|
|
|
def upsert_live_session(
|
|
self,
|
|
session_id: str,
|
|
*,
|
|
kind: str | None = None,
|
|
started_at_utc: datetime | str | None = None,
|
|
ended_at_utc: datetime | str | None = None,
|
|
status: str | None = None,
|
|
duration_ms: int | None = None,
|
|
source: str | None = None,
|
|
payload: Mapping[str, Any] | None = None,
|
|
) -> bool:
|
|
if not session_id:
|
|
return False
|
|
fields: dict[str, Any] = {"session_id": session_id}
|
|
if kind is not None:
|
|
fields["kind"] = kind
|
|
if started_at_utc is not None:
|
|
fields["started_at_utc"] = started_at_utc
|
|
if ended_at_utc is not None:
|
|
fields["ended_at_utc"] = ended_at_utc
|
|
if status is not None:
|
|
fields["status"] = status
|
|
if duration_ms is not None:
|
|
fields["duration_ms"] = max(0, int(duration_ms))
|
|
if source is not None:
|
|
fields["source"] = source
|
|
if payload is not None:
|
|
fields["payload_json"] = _json(payload)
|
|
if not self._started or self._closing:
|
|
self._dropped += 1
|
|
return False
|
|
try:
|
|
self.queue.put_nowait(("__live_session_upsert__", _clean_fields(fields)))
|
|
return True
|
|
except Exception as exc:
|
|
self._dropped += 1
|
|
self._last_error = f"{type(exc).__name__}: {exc}"
|
|
return False
|
|
|
|
def begin_live_session(
|
|
self,
|
|
session_id: str,
|
|
*,
|
|
kind: str = "live",
|
|
started_at_utc: datetime | str | None = None,
|
|
status: str = "running",
|
|
source: str | None = None,
|
|
payload: Mapping[str, Any] | None = None,
|
|
) -> bool:
|
|
return self.upsert_live_session(
|
|
session_id, kind=kind, started_at_utc=started_at_utc or utc_now(),
|
|
status=status, source=source, payload=payload,
|
|
)
|
|
|
|
def end_live_session(
|
|
self,
|
|
session_id: str,
|
|
*,
|
|
ended_at_utc: datetime | str | None = None,
|
|
status: str = "ended",
|
|
duration_ms: int | None = None,
|
|
payload: Mapping[str, Any] | None = None,
|
|
) -> bool:
|
|
return self.upsert_live_session(
|
|
session_id, ended_at_utc=ended_at_utc or utc_now(), status=status,
|
|
duration_ms=duration_ms, payload=payload,
|
|
)
|
|
|
|
def upsert_user_snapshot(
|
|
self,
|
|
platform: str,
|
|
platform_user_id: str,
|
|
*,
|
|
display_name: str | None = None,
|
|
avatar_url: str | None = None,
|
|
user_level: int | None = None,
|
|
is_admin: bool = False,
|
|
snapshot: Mapping[str, Any] | None = None,
|
|
seen_at_utc: str | None = None,
|
|
) -> bool:
|
|
if not self._started or self._closing:
|
|
return False
|
|
seen = seen_at_utc or utc_now()
|
|
fields = _clean_fields({
|
|
"platform": platform, "platform_user_id": platform_user_id, "display_name": display_name,
|
|
"avatar_url": avatar_url, "user_level": user_level, "is_admin": is_admin,
|
|
"first_seen_at_utc": seen, "last_seen_at_utc": seen, "snapshot_json": _json(snapshot or {}),
|
|
})
|
|
try:
|
|
self.queue.put_nowait(("__user_upsert__", fields))
|
|
return True
|
|
except Exception:
|
|
self._dropped += 1
|
|
return False
|
|
|
|
def load_user_points(self, platform: str) -> dict[str, int]:
|
|
"""同步读取某平台的用户积分（权威余额）。仅可在事件循环线程调用。"""
|
|
connection = self._connection
|
|
if not self._started or connection is None:
|
|
return {}
|
|
rows = connection.execute(
|
|
"SELECT platform_user_id, points FROM users WHERE platform=?",
|
|
(platform,),
|
|
).fetchall()
|
|
return {str(row[0]): int(row[1]) for row in rows}
|
|
|
|
async def set_user_points(self, platform: str, platform_user_id: str, points: int) -> bool:
|
|
"""权威写入用户积分：串行执行并等待落盘后才返回。"""
|
|
if not self._started or self._closing:
|
|
return False
|
|
try:
|
|
await self.execute(self._set_user_points_impl, platform, platform_user_id, points)
|
|
return True
|
|
except Exception as exc:
|
|
self._last_error = f"{type(exc).__name__}: {exc}"
|
|
return False
|
|
|
|
@staticmethod
|
|
def _set_user_points_impl(connection, platform, platform_user_id, points):
|
|
now = utc_now()
|
|
connection.execute(
|
|
"""INSERT INTO users (
|
|
platform, platform_user_id, display_name, first_seen_at_utc, last_seen_at_utc, points
|
|
) VALUES (?, ?, NULL, ?, ?, ?)
|
|
ON CONFLICT(platform, platform_user_id) DO UPDATE SET
|
|
points=excluded.points,
|
|
last_seen_at_utc=MAX(users.last_seen_at_utc, excluded.last_seen_at_utc)""",
|
|
(platform, platform_user_id, now, now, int(points)),
|
|
)
|
|
connection.commit()
|
|
|
|
def record_signin(self, event_id: str, *, platform: str, platform_user_id: str, points_awarded: int = 0, **fields: Any) -> bool:
|
|
return self._record_timed("signin_events", "signed_at_utc", event_id=event_id, platform=platform, platform_user_id=platform_user_id, points_awarded=points_awarded, **fields)
|
|
|
|
def record_point_transaction(self, transaction_id: str, amount: int, **fields: Any) -> bool:
|
|
return self._record_timed("point_transactions", "occurred_at_utc", transaction_id=transaction_id, amount=amount, **fields)
|
|
|
|
def record_queue_event(self, queue_name: str, action: str, **fields: Any) -> bool:
|
|
fields.setdefault("event_id", self.make_event_id(queue_name, action, utc_now(), fields))
|
|
return self._record_timed("queue_events", "occurred_at_utc", queue_name=queue_name, action=action, **fields)
|
|
|
|
def record_login_session(self, login_session_id: str, **fields: Any) -> bool:
|
|
return self._record_timed("login_sessions", "started_at_utc", login_session_id=login_session_id, **fields)
|
|
|
|
def record_group_run(self, group_run_id: str, **fields: Any) -> bool:
|
|
return self._record_timed("group_runs", "started_at_utc", group_run_id=group_run_id, **fields)
|
|
|
|
def record_song_request(self, request_id: str, **fields: Any) -> bool:
|
|
return self._record_timed("song_requests", "requested_at_utc", request_id=request_id, **fields)
|
|
|
|
def record_playback_session(self, playback_id: str, **fields: Any) -> bool:
|
|
return self._record_timed("playback_sessions", "started_at_utc", playback_id=playback_id, **fields)
|
|
|
|
def record_gift(self, event_id: str, **fields: Any) -> bool:
|
|
return self._record_timed("gift_events", "occurred_at_utc", event_id=event_id, **fields)
|
|
|
|
def record_broadcast_request(self, request_id: str, **fields: Any) -> bool:
|
|
return self._record_timed("broadcast_requests", "requested_at_utc", request_id=request_id, **fields)
|
|
|
|
def record_tts_request(self, request_id: str, **fields: Any) -> bool:
|
|
return self._record_timed("tts_requests", "requested_at_utc", request_id=request_id, **fields)
|
|
|
|
def record_bilibili_connection(self, connection_id: str, **fields: Any) -> bool:
|
|
return self._record_timed("__bilibili_connection__", "connected_at_utc", connection_id=connection_id, **fields)
|
|
|
|
def record_danmu(self, event_id: str, **fields: Any) -> bool:
|
|
return self._record_timed("danmu_events", "occurred_at_utc", event_id=event_id, **fields)
|
|
|
|
def record_service_state(self, service_name: str, new_state: str, **fields: Any) -> bool:
|
|
fields.setdefault("event_id", self.make_event_id(service_name, new_state, utc_now(), fields))
|
|
return self._record_timed("service_state_events", "occurred_at_utc", service_name=service_name, new_state=new_state, **fields)
|
|
|
|
def record_admin_audit(self, event_id: str, action: str, **fields: Any) -> bool:
|
|
if fields.get("remote_address"):
|
|
fields["remote_address_hash"] = hashlib.sha256(str(fields.pop("remote_address")).encode()).hexdigest()
|
|
return self._record_timed("admin_audit_events", "occurred_at_utc", event_id=event_id, action=action, **fields)
|
|
|
|
def _record_timed(self, table: str, time_column: str, **fields: Any) -> bool:
|
|
timestamp = _utc_iso(fields.get(time_column) or utc_now())
|
|
fields[time_column] = timestamp
|
|
fields.setdefault("business_date", business_date(timestamp))
|
|
payload = fields.pop("payload", None)
|
|
if payload is not None:
|
|
fields["payload_json"] = _json(payload)
|
|
return self._enqueue_record(table, fields)
|
|
|
|
async def import_once(
|
|
self,
|
|
source_key: str,
|
|
source_type: str,
|
|
records: Sequence[tuple[str, Mapping[str, Any]]],
|
|
*,
|
|
fingerprint: str | None = None,
|
|
checkpoint_key: str = "default",
|
|
checkpoint_value: str | None = None,
|
|
metadata: Mapping[str, Any] | None = None,
|
|
) -> bool:
|
|
"""幂等导入：相同 source_key+fingerprint 成功导入后不重复执行。"""
|
|
if not self._started or self._closing:
|
|
return False
|
|
loop = asyncio.get_running_loop()
|
|
completed = loop.create_future()
|
|
try:
|
|
await self.queue.put(("__import__", {
|
|
"source_key": source_key, "source_type": source_type, "fingerprint": fingerprint,
|
|
"records": [(table, _clean_fields(fields)) for table, fields in records],
|
|
"checkpoint_key": checkpoint_key, "checkpoint_value": checkpoint_value,
|
|
"metadata_json": _json(metadata or {}), "future": completed,
|
|
}))
|
|
return bool(await completed)
|
|
except Exception as exc:
|
|
self._last_error = f"{type(exc).__name__}: {exc}"
|
|
return False
|
|
|
|
async def query_daily_statistics(self, start_date: str | None = None, end_date: str | None = None) -> list[dict[str, Any]]:
|
|
"""查询每日统计视图；查询失败返回空列表。"""
|
|
try:
|
|
return await self.query_daily_statistics_strict(start_date, end_date)
|
|
except Exception as exc:
|
|
self._last_error = f"{type(exc).__name__}: {exc}"
|
|
return []
|
|
|
|
async def query_daily_statistics_strict(
|
|
self,
|
|
start_date: str | None = None,
|
|
end_date: str | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
"""查询每日统计视图；查询失败时向调用方抛出异常。"""
|
|
await self.flush()
|
|
rows = await asyncio.to_thread(self._query_daily_sync, start_date, end_date)
|
|
self._last_error = None
|
|
return rows
|
|
|
|
def runtime_snapshot(self) -> dict[str, Any]:
|
|
"""返回不执行磁盘完整性检查的轻量运行状态。"""
|
|
return {
|
|
"started": self._started,
|
|
"closing": self._closing,
|
|
"writer_alive": bool(self._writer_task and not self._writer_task.done()),
|
|
"queue_size": self.queue.qsize(),
|
|
"queue_capacity": self.queue.maxsize,
|
|
"last_error": self._last_error,
|
|
}
|
|
|
|
def _query_daily_sync(self, start_date: str | None, end_date: str | None) -> list[dict[str, Any]]:
|
|
clauses: list[str] = []
|
|
params: list[str] = []
|
|
if start_date:
|
|
clauses.append("business_date >= ?")
|
|
params.append(start_date)
|
|
if end_date:
|
|
clauses.append("business_date <= ?")
|
|
params.append(end_date)
|
|
where = f" WHERE {' AND '.join(clauses)}" if clauses else ""
|
|
uri = f"file:{self.database_path.as_posix()}?mode=ro"
|
|
connection = sqlite3.connect(uri, uri=True, timeout=5.0)
|
|
try:
|
|
connection.row_factory = sqlite3.Row
|
|
connection.execute("PRAGMA busy_timeout=5000")
|
|
cursor = connection.execute(f"SELECT * FROM daily_live_statistics{where} ORDER BY business_date", params)
|
|
rows = [dict(row) for row in cursor]
|
|
columns = [item[0] for item in cursor.description or ()]
|
|
|
|
live_rows = connection.execute(
|
|
"SELECT kind, started_at_utc, ended_at_utc FROM live_sessions WHERE ended_at_utc IS NOT NULL"
|
|
).fetchall()
|
|
duration_maps = {
|
|
"live_duration_ms": self._merged_daily_interval_durations([
|
|
(row[1], row[2]) for row in live_rows if row[0] in {"live", "stream", "streaming"}
|
|
]),
|
|
"process_duration_ms": self._merged_daily_interval_durations([
|
|
(row[1], row[2]) for row in live_rows if row[0] == "process"
|
|
]),
|
|
"bilibili_connection_duration_ms": self._merged_daily_interval_durations([
|
|
(row[1], row[2]) for row in live_rows if row[0] == "bilibili_connection"
|
|
]),
|
|
"group_run_total_duration_ms": self._merged_daily_interval_durations([
|
|
(row[0], row[1]) for row in connection.execute(
|
|
"SELECT started_at_utc, ended_at_utc FROM group_runs WHERE ended_at_utc IS NOT NULL"
|
|
).fetchall()
|
|
]),
|
|
}
|
|
|
|
by_day = {str(row["business_date"]): row for row in rows}
|
|
duration_days = set().union(*(values.keys() for values in duration_maps.values()))
|
|
for day in duration_days:
|
|
if (start_date and day < start_date) or (end_date and day > end_date):
|
|
continue
|
|
if day not in by_day:
|
|
by_day[day] = {column: 0 for column in columns}
|
|
by_day[day]["business_date"] = day
|
|
for day, row in by_day.items():
|
|
for column, values in duration_maps.items():
|
|
row[column] = values.get(day, 0)
|
|
return [by_day[day] for day in sorted(by_day)]
|
|
finally:
|
|
connection.close()
|
|
|
|
async def health_snapshot(self) -> dict[str, Any]:
|
|
snapshot: dict[str, Any] = {
|
|
"started": self._started,
|
|
"closing": self._closing,
|
|
"writer_alive": bool(self._writer_task and not self._writer_task.done()),
|
|
"queue_size": self.queue.qsize(),
|
|
"queue_capacity": self.queue.maxsize,
|
|
"written": self._written,
|
|
"dropped": self._dropped,
|
|
"last_error": self._last_error,
|
|
"database_path": str(self.database_path),
|
|
"process_session_id": self._process_session_id,
|
|
}
|
|
if self._started:
|
|
try:
|
|
await self.flush()
|
|
snapshot["database"] = await asyncio.to_thread(self._health_sync)
|
|
except Exception as exc:
|
|
snapshot["database"] = {"ok": False, "error": f"{type(exc).__name__}: {exc}"}
|
|
else:
|
|
snapshot["database"] = {"ok": False, "error": "not_started"}
|
|
return snapshot
|
|
|
|
def _health_sync(self) -> dict[str, Any]:
|
|
uri = f"file:{self.database_path.as_posix()}?mode=ro"
|
|
connection = sqlite3.connect(uri, uri=True, timeout=5.0)
|
|
try:
|
|
result = connection.execute("PRAGMA quick_check").fetchone()
|
|
return {
|
|
"ok": bool(result and result[0] == "ok"),
|
|
"quick_check": result[0] if result else None,
|
|
"size_bytes": self.database_path.stat().st_size if self.database_path.exists() else 0,
|
|
"schema_version": connection.execute("PRAGMA user_version").fetchone()[0],
|
|
}
|
|
finally:
|
|
connection.close()
|