1437 lines
68 KiB
Python
1437 lines
68 KiB
Python
"""
|
||
数据库存储模块
|
||
- MySQL 异步连接池
|
||
- 积分榜、赛程、比赛详情等数据 UPSERT
|
||
"""
|
||
import json
|
||
from datetime import datetime
|
||
from typing import Dict, Any, List, Optional
|
||
|
||
import hashlib
|
||
|
||
import aiomysql
|
||
import redis.asyncio as aioredis
|
||
from loguru import logger
|
||
|
||
from src.core.config import get_config
|
||
|
||
|
||
class Database:
|
||
"""MySQL 异步数据库管理"""
|
||
|
||
def __init__(self):
|
||
self._cfg = get_config().database
|
||
self._redis_cfg = get_config().redis
|
||
self._pool: Optional[aiomysql.Pool] = None
|
||
self._redis: Optional[aioredis.Redis] = None
|
||
self._prefix = self._cfg.prefix
|
||
|
||
async def _get_redis(self) -> aioredis.Redis:
|
||
if self._redis is None:
|
||
self._redis = aioredis.Redis(
|
||
host=self._redis_cfg.host,
|
||
port=self._redis_cfg.port,
|
||
password=self._redis_cfg.password or None,
|
||
db=self._redis_cfg.db,
|
||
decode_responses=True,
|
||
)
|
||
return self._redis
|
||
|
||
async def connect(self):
|
||
if self._pool:
|
||
return
|
||
self._pool = await aiomysql.create_pool(
|
||
host=self._cfg.host,
|
||
port=self._cfg.port,
|
||
user=self._cfg.username,
|
||
password=self._cfg.password,
|
||
db=self._cfg.database,
|
||
charset=self._cfg.charset,
|
||
autocommit=True,
|
||
minsize=2,
|
||
maxsize=self._cfg.pool_size,
|
||
)
|
||
logger.info(f"数据库连接池已创建: {self._cfg.host}:{self._cfg.port}/{self._cfg.database}")
|
||
|
||
async def close(self):
|
||
if self._redis:
|
||
await self._redis.close()
|
||
self._redis = None
|
||
if self._pool:
|
||
self._pool.close()
|
||
await self._pool.wait_closed()
|
||
self._pool = None
|
||
|
||
async def execute(self, sql: str, args=None) -> int:
|
||
await self.connect()
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
await cur.execute(sql, args)
|
||
return cur.rowcount
|
||
|
||
async def fetchall(self, sql: str, args=None) -> List[Dict]:
|
||
await self.connect()
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor(aiomysql.DictCursor) as cur:
|
||
await cur.execute(sql, args)
|
||
return await cur.fetchall()
|
||
|
||
async def fetchone(self, sql: str, args=None) -> Optional[Dict]:
|
||
await self.connect()
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor(aiomysql.DictCursor) as cur:
|
||
await cur.execute(sql, args)
|
||
return await cur.fetchone()
|
||
|
||
# ── 积分榜 ──
|
||
|
||
async def upsert_standings(self, standings: List[Dict[str, Any]]) -> int:
|
||
if not standings:
|
||
return 0
|
||
await self.connect()
|
||
count = 0
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
for s in standings:
|
||
await cur.execute(f"""
|
||
INSERT INTO `{self._prefix}standings`
|
||
(season_id, round_number, team_id, team_name, team_logo,
|
||
`rank`, points, played, won, drawn, lost,
|
||
goals_for, goals_against, goal_diff,
|
||
recent_form, deduction_points, deduction_reason,
|
||
created_at, updated_at)
|
||
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,NOW(),NOW())
|
||
ON DUPLICATE KEY UPDATE
|
||
team_name=VALUES(team_name), team_logo=VALUES(team_logo),
|
||
`rank`=VALUES(`rank`), points=VALUES(points),
|
||
played=VALUES(played), won=VALUES(won), drawn=VALUES(drawn), lost=VALUES(lost),
|
||
goals_for=VALUES(goals_for), goals_against=VALUES(goals_against),
|
||
goal_diff=VALUES(goal_diff), recent_form=VALUES(recent_form),
|
||
deduction_points=VALUES(deduction_points), deduction_reason=VALUES(deduction_reason),
|
||
updated_at=NOW()
|
||
""", (
|
||
s["season_id"], s["round_number"], s["team_id"],
|
||
s["team_name"], s.get("team_logo", ""),
|
||
s["rank"], s["points"], s["played"],
|
||
s["won"], s["drawn"], s["lost"],
|
||
s["goals_for"], s["goals_against"], s["goal_diff"],
|
||
s.get("recent_form", ""),
|
||
s.get("deduction_points", 0), s.get("deduction_reason", ""),
|
||
))
|
||
count += 1
|
||
logger.info(f"积分榜 UPSERT: {count} 条")
|
||
return count
|
||
|
||
# ── 赛程/比赛 ──
|
||
|
||
_STATUS_MAP = {"fixture": 0, "live": 1, "finished": 2, "postponed": 3, "cancelled": 3}
|
||
|
||
async def upsert_matches(self, matches: List[Dict[str, Any]], league_name: str = "") -> int:
|
||
if not matches:
|
||
return 0
|
||
await self.connect()
|
||
now_ts = int(datetime.now().timestamp())
|
||
today_start = int(datetime.now().replace(hour=0, minute=0, second=0, microsecond=0).timestamp())
|
||
count = 0
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
for m in matches:
|
||
status_int = self._STATUS_MAP.get(m.get("status", "fixture"), 0)
|
||
md = m.get("match_date")
|
||
match_time = int(md.timestamp()) if md else 0
|
||
if match_time < today_start:
|
||
continue
|
||
home_score = m.get("home_score")
|
||
away_score = m.get("away_score")
|
||
home_score = int(home_score) if home_score is not None and home_score != "" else 0
|
||
away_score = int(away_score) if away_score is not None and away_score != "" else 0
|
||
half_score = str(m.get("halftime_score", "")) if m.get("halftime_score") else ""
|
||
lg_name = league_name or m.get("competition_name", "")
|
||
|
||
await cur.execute(f"""
|
||
INSERT INTO `{self._prefix}match`
|
||
(match_id, competition_id,
|
||
home_team_id, away_team_id,
|
||
league_name, round_name,
|
||
home_team, home_icon, home_score,
|
||
away_team, away_icon, away_score,
|
||
sport_type, status, match_time, half_score,
|
||
is_show, create_time, update_time)
|
||
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,1,%s,%s,%s,1,%s,%s)
|
||
ON DUPLICATE KEY UPDATE
|
||
status=VALUES(status),
|
||
home_score=IF(VALUES(status)=0, home_score, VALUES(home_score)),
|
||
away_score=IF(VALUES(status)=0, away_score, VALUES(away_score)),
|
||
half_score=IF(VALUES(half_score)='', half_score, VALUES(half_score)),
|
||
home_team=VALUES(home_team), away_team=VALUES(away_team),
|
||
home_icon=IF(VALUES(home_icon)='', home_icon, VALUES(home_icon)),
|
||
away_icon=IF(VALUES(away_icon)='', away_icon, VALUES(away_icon)),
|
||
match_time=IF(VALUES(match_time)=0, match_time, VALUES(match_time)),
|
||
update_time=VALUES(update_time)
|
||
""", (
|
||
m["match_id"], m.get("competition_id", 0),
|
||
m.get("home_team_id", ""), m.get("away_team_id", ""),
|
||
lg_name, m.get("round_name", ""),
|
||
m.get("home_team_name", ""), m.get("home_team_logo", ""), home_score,
|
||
m.get("away_team_name", ""), m.get("away_team_logo", ""), away_score,
|
||
status_int, match_time, half_score,
|
||
now_ts, now_ts,
|
||
))
|
||
count += 1
|
||
logger.info(f"赛程 UPSERT: {count} 条")
|
||
return count
|
||
|
||
# ── 比赛类型配置 ──
|
||
|
||
async def upsert_match_types(self, types: List[Dict[str, Any]]) -> int:
|
||
if not types:
|
||
return 0
|
||
await self.connect()
|
||
count = 0
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
for t in types:
|
||
await cur.execute(f"""
|
||
INSERT INTO `{self._prefix}match_type_config`
|
||
(id, label, type, sort, api, created_at, updated_at)
|
||
VALUES (%s,%s,%s,%s,%s,NOW(),NOW())
|
||
ON DUPLICATE KEY UPDATE
|
||
label=VALUES(label), type=VALUES(type),
|
||
sort=VALUES(sort), api=VALUES(api), updated_at=NOW()
|
||
""", (t["id"], t["label"], t["type"], t["sort"], t["api"]))
|
||
count += 1
|
||
logger.info(f"比赛类型 UPSERT: {count} 条")
|
||
return count
|
||
|
||
# ── 比赛(la_match) ──
|
||
|
||
STATUS_MAP = {"Fixture": 0, "Playing": 1, "Played": 2}
|
||
|
||
async def upsert_la_match(self, items: List[Dict[str, Any]]) -> int:
|
||
if not items:
|
||
return 0
|
||
await self.connect()
|
||
count = 0
|
||
now_ts = int(datetime.now().timestamp())
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
for m in items:
|
||
match_id = int(m["match_id"])
|
||
competition_id = int(m.get("competition_id") or 0)
|
||
home_team_id = str(m.get("team_A_id", ""))
|
||
away_team_id = str(m.get("team_B_id", ""))
|
||
status = self.STATUS_MAP.get(m.get("status", ""), 0)
|
||
start_play = m.get("start_play", "")
|
||
match_time = int(m.get("sort_timestamp") or 0)
|
||
if not match_time and start_play:
|
||
try:
|
||
match_time = int(datetime.strptime(start_play, "%Y-%m-%d %H:%M:%S").timestamp())
|
||
except Exception:
|
||
pass
|
||
home_score = int(m.get("fs_A") or 0) if m.get("fs_A") else 0
|
||
away_score = int(m.get("fs_B") or 0) if m.get("fs_B") else 0
|
||
hts_a = m.get("hts_A", "")
|
||
hts_b = m.get("hts_B", "")
|
||
half_score = f"{hts_a}-{hts_b}" if hts_a and hts_b else ""
|
||
cmp_type = m.get("cmp_type", "soccer")
|
||
sport_type_map = {"soccer": 1, "basketball": 2, "tennis": 3, "esport": 4, "synthesize": 5}
|
||
sport_type = sport_type_map.get(cmp_type, 1)
|
||
home_odds = float(m.get("home") or 0)
|
||
draw_odds = float(m.get("draw") or 0)
|
||
away_odds = float(m.get("away") or 0)
|
||
await cur.execute(f"""
|
||
INSERT INTO `{self._prefix}match`
|
||
(match_id, competition_id, home_team_id, away_team_id,
|
||
league_name, round_name, league_icon,
|
||
home_team, home_icon, home_score,
|
||
away_team, away_icon, away_score,
|
||
sport_type, status, match_time,
|
||
current_minute, half_score,
|
||
home_odds, draw_odds, away_odds,
|
||
home_corner, away_corner,
|
||
home_yellow, away_yellow, home_red, away_red,
|
||
is_hot, is_show, sort,
|
||
create_time, update_time)
|
||
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
|
||
ON DUPLICATE KEY UPDATE
|
||
status=VALUES(status),
|
||
home_score=VALUES(home_score), away_score=VALUES(away_score),
|
||
half_score=VALUES(half_score), current_minute=VALUES(current_minute),
|
||
home_odds=VALUES(home_odds), draw_odds=VALUES(draw_odds), away_odds=VALUES(away_odds),
|
||
home_corner=VALUES(home_corner), away_corner=VALUES(away_corner),
|
||
home_yellow=VALUES(home_yellow), away_yellow=VALUES(away_yellow),
|
||
home_red=VALUES(home_red), away_red=VALUES(away_red),
|
||
update_time=VALUES(update_time)
|
||
""", (
|
||
match_id, competition_id, home_team_id, away_team_id,
|
||
m.get("competition_name", "")[:100],
|
||
m.get("round_name", "")[:50],
|
||
m.get("bk_logo", "")[:255],
|
||
m.get("team_A_name", "")[:100],
|
||
m.get("team_A_logo", "")[:255],
|
||
home_score,
|
||
m.get("team_B_name", "")[:100],
|
||
m.get("team_B_logo", "")[:255],
|
||
away_score,
|
||
sport_type, status, match_time,
|
||
m.get("minute", "")[:20] if m.get("minute") else "",
|
||
half_score[:50],
|
||
home_odds, draw_odds, away_odds,
|
||
int(m.get("corner_A") or 0), int(m.get("corner_B") or 0),
|
||
int(m.get("yc_A") or 0), int(m.get("yc_B") or 0),
|
||
int(m.get("rc_A") or 0), int(m.get("rc_B") or 0),
|
||
int(m.get("key_match") or 0), 1, 0,
|
||
now_ts, now_ts,
|
||
))
|
||
count += 1
|
||
logger.info(f"la_match UPSERT: {count} 条")
|
||
return count
|
||
|
||
async def get_existing_match_dates(self) -> set:
|
||
await self.connect()
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
await cur.execute(f"""
|
||
SELECT DISTINCT DATE(FROM_UNIXTIME(match_time)) AS d
|
||
FROM `{self._prefix}match`
|
||
WHERE match_id > 0 AND match_time > 0
|
||
""")
|
||
rows = await cur.fetchall()
|
||
return {str(r[0]) for r in rows if r[0]}
|
||
|
||
async def get_existing_match_ids(self) -> set:
|
||
await self.connect()
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
await cur.execute(f"SELECT match_id FROM `{self._prefix}match` WHERE match_id > 0")
|
||
rows = await cur.fetchall()
|
||
return {int(r[0]) for r in rows}
|
||
|
||
# ── 实时比赛查询/更新 ──
|
||
|
||
async def get_live_matches(self, now_ts: int, lookback: int = 86400) -> List[Dict]:
|
||
"""查询已到开赛时间但未结束的比赛 (status IN(0,1) 且 match_time 在 lookback 秒内)"""
|
||
await self.connect()
|
||
earliest = now_ts - lookback
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor(aiomysql.DictCursor) as cur:
|
||
await cur.execute(f"""
|
||
SELECT id, match_id, home_team, away_team, league_name,
|
||
match_time, status, home_score, away_score, sport_type
|
||
FROM `{self._prefix}match`
|
||
WHERE match_id > 0
|
||
AND match_time > 0
|
||
AND match_time <= %s
|
||
AND match_time >= %s
|
||
AND status IN (0, 1)
|
||
ORDER BY match_time ASC
|
||
""", (now_ts, earliest))
|
||
return await cur.fetchall()
|
||
|
||
async def update_match_live(
|
||
self,
|
||
match_id: int,
|
||
status: int,
|
||
home_score: int,
|
||
away_score: int,
|
||
half_score: str = "",
|
||
current_minute: str = "",
|
||
):
|
||
"""更新比赛实时比分和状态"""
|
||
await self.connect()
|
||
now_ts = int(datetime.now().timestamp())
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
await cur.execute(f"""
|
||
UPDATE `{self._prefix}match`
|
||
SET status=%s, home_score=%s, away_score=%s,
|
||
half_score=%s, current_minute=%s, update_time=%s
|
||
WHERE match_id=%s
|
||
""", (status, home_score, away_score, half_score, current_minute, now_ts, match_id))
|
||
logger.debug(f"la_match 更新: match_id={match_id} status={status} {home_score}-{away_score}")
|
||
|
||
# 允许动态更新的技术统计列白名单
|
||
_ALLOWED_STAT_COLS = {
|
||
"tech_stats",
|
||
}
|
||
|
||
async def update_match_stats(self, match_id: int, updates: Dict[str, str]):
|
||
"""动态更新 la_match 的技术统计字段"""
|
||
safe = {k: v for k, v in updates.items() if k in self._ALLOWED_STAT_COLS}
|
||
if not safe:
|
||
return
|
||
await self.connect()
|
||
now_ts = int(datetime.now().timestamp())
|
||
set_clauses = ", ".join(f"`{col}`=%s" for col in safe)
|
||
set_clauses += ", update_time=%s"
|
||
vals = list(safe.values()) + [now_ts, match_id]
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
await cur.execute(
|
||
f"UPDATE `{self._prefix}match` SET {set_clauses} WHERE match_id=%s",
|
||
vals,
|
||
)
|
||
logger.info(f"la_match 技术统计更新: match_id={match_id} fields={list(safe.keys())}")
|
||
|
||
async def get_league_sessions(self) -> List[Dict[str, Any]]:
|
||
"""从 la_league 获取所有配置了 sessionid 或 api 的联赛"""
|
||
await self.connect()
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor(aiomysql.DictCursor) as cur:
|
||
await cur.execute(f"""
|
||
SELECT id, label, sessionid, league_id, sport_type, api
|
||
FROM `{self._prefix}league`
|
||
WHERE `type` = 'league'
|
||
AND is_show = 1
|
||
AND (
|
||
(sessionid IS NOT NULL AND sessionid != '')
|
||
OR (api IS NOT NULL AND api != '')
|
||
)
|
||
""")
|
||
return await cur.fetchall()
|
||
|
||
async def upsert_match_rounds(self, rounds: List[Dict[str, Any]]) -> int:
|
||
"""批量插入/更新轮次到 la_match_round"""
|
||
if not rounds:
|
||
return 0
|
||
await self.connect()
|
||
count = 0
|
||
now_ts = int(datetime.now().timestamp())
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
for r in rounds:
|
||
await cur.execute(f"""
|
||
INSERT INTO `{self._prefix}match_round`
|
||
(league_id, season_id, round_id, round_name, gameweek, url, create_time)
|
||
VALUES (%s, %s, %s, %s, %s, %s, %s)
|
||
ON DUPLICATE KEY UPDATE
|
||
league_id=VALUES(league_id),
|
||
round_name=VALUES(round_name),
|
||
url=VALUES(url)
|
||
""", (
|
||
r['league_id'], r['season_id'], r['round_id'],
|
||
r['round_name'], r['gameweek'], r['url'], now_ts,
|
||
))
|
||
count += 1
|
||
logger.info(f"la_match_round UPSERT: {count} 条")
|
||
return count
|
||
|
||
async def check_round_skip(self, round_name: str, league_name: str) -> bool:
|
||
"""检查某轮次是否可以跳过（查 la_match）：
|
||
- 已有数据 且 全部已结束(status=2)
|
||
- 已有数据 且 全部比赛时间在未来（还未开始）
|
||
"""
|
||
await self.connect()
|
||
now_ts = int(datetime.now().timestamp())
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor(aiomysql.DictCursor) as cur:
|
||
await cur.execute(f"""
|
||
SELECT COUNT(*) AS total,
|
||
SUM(CASE WHEN `status` = 2 THEN 1 ELSE 0 END) AS finished,
|
||
SUM(CASE WHEN match_time > %s THEN 1 ELSE 0 END) AS future
|
||
FROM `{self._prefix}match`
|
||
WHERE round_name = %s AND league_name = %s
|
||
""", (now_ts, round_name, league_name))
|
||
row = await cur.fetchone()
|
||
if not row or row['total'] == 0:
|
||
return False
|
||
total = int(row['total'])
|
||
finished = int(row['finished'] or 0)
|
||
future = int(row['future'] or 0)
|
||
return finished == total or future == total
|
||
|
||
_SCHEDULE_STATUS_MAP = {"Fixture": 0, "Playing": 1, "Played": 2}
|
||
|
||
@staticmethod
|
||
def _safe_int(val, default: int = 0) -> int:
|
||
if val is None:
|
||
return default
|
||
try:
|
||
return int(str(val).strip())
|
||
except (ValueError, TypeError):
|
||
return default
|
||
|
||
async def upsert_match_data_batch(self, items: List[Dict[str, Any]], league_name: str = "", sport_type: int = 1) -> Dict[str, int]:
|
||
"""批量插入/更新比赛数据到 la_match（来自 schedule API 或 tab API 的 matches）"""
|
||
if not items:
|
||
return {"inserted": 0, "updated": 0, "unchanged": 0, "total": 0}
|
||
await self.connect()
|
||
inserted = 0
|
||
updated = 0
|
||
unchanged = 0
|
||
now_ts = int(datetime.now().timestamp())
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
for m in items:
|
||
match_id = m.get("match_id", "")
|
||
if not match_id:
|
||
continue
|
||
match_id = self._safe_int(match_id)
|
||
if not match_id:
|
||
continue
|
||
competition_id = self._safe_int(m.get("competition_id"))
|
||
status = self._SCHEDULE_STATUS_MAP.get(m.get("status", ""), 0)
|
||
start_play = m.get("start_play", "")
|
||
match_time = 0
|
||
if start_play:
|
||
try:
|
||
match_time = int(datetime.strptime(start_play, "%Y-%m-%d %H:%M:%S").timestamp())
|
||
except Exception:
|
||
pass
|
||
home_score = self._safe_int(m.get("fs_A"))
|
||
away_score = self._safe_int(m.get("fs_B"))
|
||
half_score = ""
|
||
round_name = str(m.get("_round_name", ""))[:50]
|
||
|
||
await cur.execute(f"""
|
||
INSERT INTO `{self._prefix}match`
|
||
(match_id, competition_id, home_team_id, away_team_id,
|
||
league_name, round_name, league_icon,
|
||
home_team, home_icon, home_score,
|
||
away_team, away_icon, away_score,
|
||
sport_type, status, match_time,
|
||
current_minute, half_score,
|
||
is_show, create_time, update_time)
|
||
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,1,%s,%s)
|
||
ON DUPLICATE KEY UPDATE
|
||
status=VALUES(status),
|
||
home_score=VALUES(home_score), away_score=VALUES(away_score),
|
||
half_score=IF(VALUES(half_score)='', half_score, VALUES(half_score)),
|
||
current_minute=VALUES(current_minute),
|
||
round_name=VALUES(round_name),
|
||
home_team=VALUES(home_team), away_team=VALUES(away_team),
|
||
home_icon=IF(VALUES(home_icon)='', home_icon, VALUES(home_icon)),
|
||
away_icon=IF(VALUES(away_icon)='', away_icon, VALUES(away_icon)),
|
||
match_time=IF(VALUES(match_time)=0, match_time, VALUES(match_time)),
|
||
update_time=VALUES(update_time)
|
||
""", (
|
||
match_id, competition_id,
|
||
str(m.get("team_A_id", "")),
|
||
str(m.get("team_B_id", "")),
|
||
league_name[:100],
|
||
round_name,
|
||
"",
|
||
str(m.get("team_A_name", ""))[:100],
|
||
str(m.get("team_A_logo", ""))[:255],
|
||
home_score,
|
||
str(m.get("team_B_name", ""))[:100],
|
||
str(m.get("team_B_logo", ""))[:255],
|
||
away_score,
|
||
sport_type, status, match_time,
|
||
str(m.get("minute", ""))[:20],
|
||
half_score,
|
||
now_ts, now_ts,
|
||
))
|
||
affected = cur.rowcount
|
||
if affected == 1:
|
||
inserted += 1
|
||
elif affected == 2:
|
||
updated += 1
|
||
else:
|
||
unchanged += 1
|
||
total = inserted + updated + unchanged
|
||
logger.info(f"la_match UPSERT (schedule): 新增 {inserted}, 更新 {updated}, 未变 {unchanged}, 共 {total} 条")
|
||
return {"inserted": inserted, "updated": updated, "unchanged": unchanged, "total": total}
|
||
|
||
async def get_live_match_ids(self) -> List[int]:
|
||
"""获取正在进行中(status=1)的比赛match_id列表"""
|
||
await self.connect()
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
await cur.execute(f"""
|
||
SELECT match_id FROM `{self._prefix}match`
|
||
WHERE `status` = 1 AND is_show = 1
|
||
""")
|
||
rows = await cur.fetchall()
|
||
return [int(r[0]) for r in rows]
|
||
|
||
async def upsert_match_live_text(self, match_id: int, items: List[Dict[str, Any]]) -> int:
|
||
"""批量插入文字直播到 la_match_live_text"""
|
||
if not items:
|
||
return 0
|
||
await self.connect()
|
||
count = 0
|
||
now_ts = int(datetime.now().timestamp())
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
for m in items:
|
||
msg_id = int(m.get("id", 0))
|
||
if not msg_id:
|
||
continue
|
||
await cur.execute(f"""
|
||
INSERT INTO `{self._prefix}match_live_text`
|
||
(match_id, msg_id, event_type, username, avatar, message, image, `timestamp`, create_time)
|
||
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
|
||
ON DUPLICATE KEY UPDATE
|
||
message=VALUES(message), image=VALUES(image)
|
||
""", (
|
||
match_id, msg_id,
|
||
str(m.get("event_type") or ""),
|
||
str(m.get("username", ""))[:50],
|
||
str(m.get("avatar", ""))[:500],
|
||
str(m.get("message", "")),
|
||
str(m.get("image", ""))[:500],
|
||
str(m.get("timestamp", ""))[:20],
|
||
now_ts,
|
||
))
|
||
count += 1
|
||
logger.info(f"la_match_live_text UPSERT: match_id={match_id}, {count} 条")
|
||
return count
|
||
|
||
async def upsert_match_data(self, match_sample: Dict[str, Any]):
|
||
"""将完整 matchSample 快照写入 la_match_data"""
|
||
await self.connect()
|
||
match_id = str(match_sample.get("match_id", ""))
|
||
if not match_id:
|
||
return
|
||
raw_json = json.dumps(match_sample, ensure_ascii=False, default=str)
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
await cur.execute(f"""
|
||
INSERT INTO `{self._prefix}match_data`
|
||
(match_id, relate_type, relate_id,
|
||
team_A_id, team_A_name, team_A_logo,
|
||
team_B_id, team_B_name, team_B_logo,
|
||
date_utc, time_utc, start_play, sort_timestamp,
|
||
status, fs_A, fs_B,
|
||
competition_id, competition_name, round_name,
|
||
minute, minute_period,
|
||
raw_data, crawl_time)
|
||
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,NOW())
|
||
ON DUPLICATE KEY UPDATE
|
||
status=VALUES(status), fs_A=VALUES(fs_A), fs_B=VALUES(fs_B),
|
||
minute=VALUES(minute), minute_period=VALUES(minute_period),
|
||
raw_data=VALUES(raw_data), crawl_time=NOW()
|
||
""", (
|
||
match_id,
|
||
str(match_sample.get("relate_type", "match")),
|
||
str(match_sample.get("relate_id", match_id)),
|
||
str(match_sample.get("team_A_id", "")),
|
||
str(match_sample.get("team_A_name", ""))[:100],
|
||
str(match_sample.get("team_A_logo", ""))[:500],
|
||
str(match_sample.get("team_B_id", "")),
|
||
str(match_sample.get("team_B_name", ""))[:100],
|
||
str(match_sample.get("team_B_logo", ""))[:500],
|
||
str(match_sample.get("date_utc", "")),
|
||
str(match_sample.get("time_utc", "")),
|
||
str(match_sample.get("start_play", "")),
|
||
int(match_sample.get("sort_timestamp") or 0),
|
||
str(match_sample.get("status", "")),
|
||
str(match_sample.get("fs_A", "")) or None,
|
||
str(match_sample.get("fs_B", "")) or None,
|
||
str(match_sample.get("competition_id", "")),
|
||
str(match_sample.get("competition_name", ""))[:100],
|
||
str(match_sample.get("round_name", ""))[:100],
|
||
str(match_sample.get("minute", "")),
|
||
str(match_sample.get("minute_period", "")),
|
||
raw_json,
|
||
))
|
||
logger.debug(f"la_match_data UPSERT: match_id={match_id}")
|
||
|
||
# ── 文章 ──
|
||
|
||
async def get_existing_article_ids(self) -> set:
|
||
await self.connect()
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
await cur.execute(f"SELECT article_id FROM `{self._prefix}article` WHERE article_id > 0")
|
||
rows = await cur.fetchall()
|
||
return {int(r[0]) for r in rows}
|
||
|
||
async def get_article_cate_map(self) -> Dict[str, int]:
|
||
"""返回 {分类名: cid} 映射"""
|
||
await self.connect()
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor(aiomysql.DictCursor) as cur:
|
||
await cur.execute(f"SELECT id, name FROM `{self._prefix}article_cate` WHERE delete_time IS NULL")
|
||
rows = await cur.fetchall()
|
||
return {r["name"]: r["id"] for r in rows}
|
||
|
||
async def upsert_articles(self, items: List[Dict[str, Any]], cid: int = 0) -> int:
|
||
if not items:
|
||
return 0
|
||
await self.connect()
|
||
count = 0
|
||
skip_count = 0
|
||
now_ts = int(datetime.now().timestamp())
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
titles = []
|
||
for a in items:
|
||
title = str(a.get("title", "") or "").strip()[:255]
|
||
if title:
|
||
titles.append(title)
|
||
|
||
existing_title_article_ids = {}
|
||
if titles:
|
||
placeholders = ",".join(["%s"] * len(titles))
|
||
await cur.execute(
|
||
f"SELECT title, article_id FROM `{self._prefix}article` WHERE cid = %s AND title IN ({placeholders})",
|
||
[cid] + titles
|
||
)
|
||
for row in await cur.fetchall():
|
||
existing_title_article_ids.setdefault(row[0], set()).add(int(row[1]))
|
||
|
||
seen_titles = set()
|
||
for a in items:
|
||
article_id = int(a.get("id", 0) or 0)
|
||
if not article_id:
|
||
continue
|
||
title = str(a.get("title", "") or "").strip()[:255]
|
||
if title:
|
||
title_article_ids = existing_title_article_ids.get(title, set())
|
||
if title in seen_titles or (title_article_ids and article_id not in title_article_ids):
|
||
skip_count += 1
|
||
continue
|
||
seen_titles.add(title)
|
||
desc = str(a.get("description", "") or "")[:255]
|
||
thumb = str(a.get("thumb", "") or "")[:500]
|
||
image = ""
|
||
imgs = a.get("match_image_list") or []
|
||
if imgs and isinstance(imgs, list) and imgs[0].get("url"):
|
||
image = str(imgs[0]["url"])[:128]
|
||
if not image and thumb:
|
||
image = thumb[:128]
|
||
is_video = 1 if a.get("is_video") or a.get("has_video") else 0
|
||
duration = str(a.get("video_time", "") or a.get("duration", "") or "")[:20]
|
||
video_url = str(a.get("video_src", "") or a.get("video_url", "") or "")[:500]
|
||
author_info = a.get("author")
|
||
if isinstance(author_info, dict):
|
||
author = str(author_info.get("name", ""))[:255]
|
||
else:
|
||
author = str(a.get("author_name", "") or "")[:255]
|
||
published_at = str(a.get("published_at", ""))[:30]
|
||
category = str(a.get("category", "") or "")[:50]
|
||
source_url = str(a.get("share", "") or "")[:500]
|
||
comment_count = int(a.get("comments_total", 0) or 0)
|
||
sort_ts = int(a.get("sort_timestamp", 0) or 0)
|
||
await cur.execute(f"""
|
||
INSERT INTO `{self._prefix}article`
|
||
(article_id, cid, title, `desc`, image, thumb, is_video,
|
||
duration, video_url,
|
||
author, published_at, category, source_url,
|
||
comment_count, sort, is_show, create_time, update_time)
|
||
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,1,%s,%s)
|
||
ON DUPLICATE KEY UPDATE
|
||
cid=IF(cid=0 AND VALUES(cid)<>0, VALUES(cid), cid),
|
||
title=IF(VALUES(title)='', title, VALUES(title)),
|
||
`desc`=IF(VALUES(`desc`)='', `desc`, VALUES(`desc`)),
|
||
image=IF(VALUES(image)='', image, VALUES(image)),
|
||
thumb=IF(VALUES(thumb)='', thumb, VALUES(thumb)),
|
||
duration=IF(VALUES(duration)='', duration, VALUES(duration)),
|
||
video_url=IF(VALUES(video_url)='', video_url, VALUES(video_url)),
|
||
category=IF(VALUES(category)='', category, VALUES(category)),
|
||
sort=IF(VALUES(sort)=0, sort, VALUES(sort)),
|
||
comment_count=VALUES(comment_count),
|
||
update_time=VALUES(update_time)
|
||
""", (
|
||
article_id, cid, title, desc, image, thumb, is_video,
|
||
duration, video_url,
|
||
author, published_at, category, source_url,
|
||
comment_count, sort_ts, now_ts, now_ts,
|
||
))
|
||
count += 1
|
||
logger.info(f"la_article UPSERT: {count} 条, 标题重复跳过: {skip_count} 条")
|
||
return count
|
||
|
||
async def get_articles_without_content(self, limit: int = 100) -> List[Dict[str, Any]]:
|
||
await self.connect()
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor(aiomysql.DictCursor) as cur:
|
||
await cur.execute(f"""
|
||
SELECT id, article_id, title, source_url
|
||
FROM `{self._prefix}article`
|
||
WHERE article_id > 0
|
||
AND (
|
||
(content IS NULL OR content = '')
|
||
OR (is_video = 1 AND (title IS NULL OR title = ''))
|
||
)
|
||
AND content_retry < 3
|
||
ORDER BY id DESC
|
||
LIMIT %s
|
||
""", (limit,))
|
||
return await cur.fetchall()
|
||
|
||
async def update_article_content(self, article_id: int, detail: dict) -> None:
|
||
await self.connect()
|
||
now_ts = int(datetime.now().timestamp())
|
||
content = detail.get("content", "")
|
||
title = detail.get("title", "")
|
||
desc = detail.get("description", "")
|
||
author = detail.get("author", "")
|
||
published_at = detail.get("published_at", "")
|
||
video_url = detail.get("video_url", "")
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
await cur.execute(f"""
|
||
UPDATE `{self._prefix}article`
|
||
SET content = %s,
|
||
title = IF(%s = '', title, %s),
|
||
`desc` = IF(%s = '', `desc`, %s),
|
||
author = IF(%s = '', author, %s),
|
||
published_at = IF(%s = '', published_at, %s),
|
||
video_url = IF(%s = '', video_url, %s),
|
||
update_time = %s
|
||
WHERE article_id = %s
|
||
""", (content,
|
||
title, title,
|
||
desc, desc,
|
||
author, author,
|
||
published_at, published_at,
|
||
video_url, video_url,
|
||
now_ts, article_id))
|
||
|
||
async def increment_content_retry(self, article_id: int) -> None:
|
||
await self.connect()
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
await cur.execute(f"""
|
||
UPDATE `{self._prefix}article`
|
||
SET content_retry = content_retry + 1
|
||
WHERE article_id = %s
|
||
""", (article_id,))
|
||
|
||
async def set_content_retry_max(self, article_id: int) -> None:
|
||
"""页面无可提取内容，直接设为最大重试次数跳过"""
|
||
await self.connect()
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
await cur.execute(f"""
|
||
UPDATE `{self._prefix}article`
|
||
SET content_retry = 3
|
||
WHERE article_id = %s
|
||
""", (article_id,))
|
||
|
||
# ── 六合彩开奖 ──
|
||
|
||
async def get_lottery_number_mapping(self, year: int) -> dict:
|
||
"""获取指定年份的号码映射（生肖/波色/五行），返回 {number: {zodiac, color, element}}"""
|
||
await self.connect()
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor(aiomysql.DictCursor) as cur:
|
||
await cur.execute(f"""
|
||
SELECT type, attr_name, numbers
|
||
FROM `{self._prefix}lottery_number_mapping`
|
||
WHERE year = %s
|
||
""", (year,))
|
||
rows = await cur.fetchall()
|
||
|
||
mapping = {} # {number: {zodiac: '', color: '', element: ''}}
|
||
for row in rows:
|
||
nums = json.loads(row['numbers']) if isinstance(row['numbers'], str) else row['numbers']
|
||
for n in nums:
|
||
if n not in mapping:
|
||
mapping[n] = {'zodiac': '', 'color': '', 'element': ''}
|
||
t = int(row['type'])
|
||
if t == 1:
|
||
mapping[n]['zodiac'] = row['attr_name']
|
||
elif t == 2:
|
||
mapping[n]['color'] = row['attr_name']
|
||
elif t == 3:
|
||
mapping[n]['element'] = row['attr_name']
|
||
return mapping
|
||
|
||
async def get_lottery_draw_exists(self, category_id: int, period: str) -> bool:
|
||
"""检查某期开奖是否已存在且已开奖(status=1)"""
|
||
await self.connect()
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
await cur.execute(f"""
|
||
SELECT id FROM `{self._prefix}lottery_draw`
|
||
WHERE category_id = %s AND period = %s AND status = 1
|
||
""", (category_id, period))
|
||
return await cur.fetchone() is not None
|
||
|
||
async def upsert_lottery_draw(self, data: dict) -> int:
|
||
"""插入或更新六合彩开奖记录"""
|
||
await self.connect()
|
||
now_ts = int(datetime.now().timestamp())
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
await cur.execute(f"""
|
||
INSERT INTO `{self._prefix}lottery_draw`
|
||
(category_id, period, draw_date, draw_time,
|
||
numbers, special_number, zodiac, elements, color,
|
||
status, is_show, create_time, update_time)
|
||
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 1, %s, %s)
|
||
ON DUPLICATE KEY UPDATE
|
||
numbers=VALUES(numbers),
|
||
special_number=VALUES(special_number),
|
||
zodiac=VALUES(zodiac),
|
||
elements=VALUES(elements),
|
||
color=VALUES(color),
|
||
draw_time=VALUES(draw_time),
|
||
status=VALUES(status),
|
||
update_time=VALUES(update_time)
|
||
""", (
|
||
data['category_id'], data['period'], data['draw_date'], data.get('draw_time'),
|
||
json.dumps(data['numbers']), str(data['special_number']),
|
||
json.dumps(data['zodiac'], ensure_ascii=False),
|
||
json.dumps(data['elements'], ensure_ascii=False),
|
||
json.dumps(data['color'], ensure_ascii=False),
|
||
data['status'], now_ts, now_ts,
|
||
))
|
||
return cur.rowcount
|
||
|
||
async def upsert_lottery_next_draw(self, data: dict) -> int:
|
||
"""插入下一期待开奖记录（status=0）"""
|
||
await self.connect()
|
||
now_ts = int(datetime.now().timestamp())
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
await cur.execute(f"""
|
||
INSERT INTO `{self._prefix}lottery_draw`
|
||
(category_id, period, draw_date, draw_time,
|
||
numbers, special_number, zodiac, elements, color,
|
||
status, is_show, create_time, update_time)
|
||
VALUES (%s, %s, %s, %s, '[]', '', '[]', '[]', '[]', 0, 1, %s, %s)
|
||
ON DUPLICATE KEY UPDATE
|
||
draw_date=IF(VALUES(draw_date) IS NULL, draw_date, VALUES(draw_date)),
|
||
draw_time=IF(VALUES(draw_time) IS NULL, draw_time, VALUES(draw_time)),
|
||
update_time=VALUES(update_time)
|
||
""", (
|
||
data['category_id'], data['period'],
|
||
data.get('draw_date'), data.get('draw_time'),
|
||
now_ts, now_ts,
|
||
))
|
||
return cur.rowcount
|
||
|
||
async def get_pending_lottery_draws(self) -> list:
|
||
"""获取所有待开奖记录（status=0），返回 [{category_id, period, draw_date, draw_time}]"""
|
||
await self.connect()
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor(aiomysql.DictCursor) as cur:
|
||
await cur.execute(f"""
|
||
SELECT category_id, period, draw_date, draw_time
|
||
FROM `{self._prefix}lottery_draw`
|
||
WHERE status = 0
|
||
ORDER BY draw_time ASC
|
||
""")
|
||
return await cur.fetchall()
|
||
|
||
async def get_latest_lottery_draw(self, category_id: int) -> dict:
|
||
"""获取某彩种最新一条已开奖记录"""
|
||
await self.connect()
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor(aiomysql.DictCursor) as cur:
|
||
await cur.execute(f"""
|
||
SELECT category_id, period, draw_date, draw_time, numbers, special_number
|
||
FROM `{self._prefix}lottery_draw`
|
||
WHERE category_id = %s AND status = 1
|
||
ORDER BY id DESC LIMIT 1
|
||
""", (category_id,))
|
||
return await cur.fetchone() or {}
|
||
|
||
# ── 爬取日志 ──
|
||
|
||
async def log_crawl(
|
||
self,
|
||
crawl_type: str,
|
||
target: str,
|
||
success: bool,
|
||
elapsed: float,
|
||
record_count: int = 0,
|
||
error: str = "",
|
||
channel: str = "",
|
||
):
|
||
await self.connect()
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
await cur.execute(f"""
|
||
INSERT INTO `{self._prefix}crawl_log`
|
||
(crawl_type, api_url, success, response_time, match_count,
|
||
error_message, user_agent, created_at)
|
||
VALUES (%s,%s,%s,%s,%s,%s,%s,NOW())
|
||
""", (crawl_type, target, success, elapsed, record_count, error, channel))
|
||
|
||
# ── 建表 DDL ──
|
||
|
||
async def ensure_tables(self):
|
||
"""确保所需数据表存在"""
|
||
await self.connect()
|
||
ddls = [
|
||
f"""CREATE TABLE IF NOT EXISTS `{self._prefix}standings` (
|
||
`id` BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
|
||
`season_id` INT NOT NULL,
|
||
`round_number` INT NOT NULL DEFAULT 0,
|
||
`team_id` INT NOT NULL,
|
||
`team_name` VARCHAR(100) NOT NULL DEFAULT '',
|
||
`team_logo` VARCHAR(500) DEFAULT '',
|
||
`rank` INT NOT NULL DEFAULT 0,
|
||
`points` INT NOT NULL DEFAULT 0,
|
||
`played` INT NOT NULL DEFAULT 0,
|
||
`won` INT NOT NULL DEFAULT 0,
|
||
`drawn` INT NOT NULL DEFAULT 0,
|
||
`lost` INT NOT NULL DEFAULT 0,
|
||
`goals_for` INT NOT NULL DEFAULT 0,
|
||
`goals_against` INT NOT NULL DEFAULT 0,
|
||
`goal_diff` INT NOT NULL DEFAULT 0,
|
||
`recent_form` VARCHAR(20) DEFAULT '',
|
||
`deduction_points` INT DEFAULT 0,
|
||
`deduction_reason` VARCHAR(500) DEFAULT '',
|
||
`created_at` DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||
`updated_at` DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
||
UNIQUE KEY `uk_season_round_team` (`season_id`, `round_number`, `team_id`),
|
||
KEY `idx_season` (`season_id`),
|
||
KEY `idx_team` (`team_id`)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci""",
|
||
|
||
f"""CREATE TABLE IF NOT EXISTS `{self._prefix}matches` (
|
||
`id` BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
|
||
`match_id` BIGINT NOT NULL,
|
||
`season_id` INT NOT NULL,
|
||
`round_number` INT DEFAULT 0,
|
||
`round_name` VARCHAR(100) DEFAULT '',
|
||
`match_date` DATETIME DEFAULT NULL,
|
||
`status` VARCHAR(20) DEFAULT 'fixture',
|
||
`home_team_id` INT DEFAULT NULL,
|
||
`home_team_name` VARCHAR(100) DEFAULT '',
|
||
`home_team_logo` VARCHAR(500) DEFAULT '',
|
||
`away_team_id` INT DEFAULT NULL,
|
||
`away_team_name` VARCHAR(100) DEFAULT '',
|
||
`away_team_logo` VARCHAR(500) DEFAULT '',
|
||
`home_score` INT DEFAULT NULL,
|
||
`away_score` INT DEFAULT NULL,
|
||
`halftime_score` VARCHAR(20) DEFAULT '',
|
||
`venue` VARCHAR(200) DEFAULT '',
|
||
`competition_id` INT DEFAULT NULL,
|
||
`competition_name` VARCHAR(100) DEFAULT '',
|
||
`created_at` DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||
`updated_at` DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
||
UNIQUE KEY `uk_match_id` (`match_id`),
|
||
KEY `idx_season` (`season_id`),
|
||
KEY `idx_status` (`status`),
|
||
KEY `idx_date` (`match_date`)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci""",
|
||
|
||
f"""CREATE TABLE IF NOT EXISTS `{self._prefix}match_type_config` (
|
||
`id` INT PRIMARY KEY,
|
||
`label` VARCHAR(100) DEFAULT '',
|
||
`type` VARCHAR(50) DEFAULT '',
|
||
`sort` INT DEFAULT 0,
|
||
`api` VARCHAR(500) DEFAULT '',
|
||
`created_at` DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||
`updated_at` DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci""",
|
||
|
||
f"""CREATE TABLE IF NOT EXISTS `{self._prefix}crawl_log` (
|
||
`id` BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
|
||
`crawl_type` VARCHAR(50) NOT NULL,
|
||
`target` VARCHAR(200) DEFAULT '',
|
||
`success` TINYINT(1) DEFAULT 0,
|
||
`elapsed` FLOAT DEFAULT 0,
|
||
`record_count` INT DEFAULT 0,
|
||
`error_message` TEXT,
|
||
`channel` VARCHAR(50) DEFAULT '',
|
||
`created_at` DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||
KEY `idx_type` (`crawl_type`),
|
||
KEY `idx_time` (`created_at`)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci""",
|
||
|
||
f"""CREATE TABLE IF NOT EXISTS `{self._prefix}crawl_error_log` (
|
||
`id` BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
|
||
`task_name` VARCHAR(100) NOT NULL DEFAULT '',
|
||
`request_url` VARCHAR(1000) NOT NULL DEFAULT '',
|
||
`request_method` VARCHAR(10) NOT NULL DEFAULT 'GET',
|
||
`request_params` TEXT,
|
||
`http_status` INT NOT NULL DEFAULT 0,
|
||
`error_type` VARCHAR(100) NOT NULL DEFAULT '',
|
||
`error_message` TEXT,
|
||
`response_body` TEXT,
|
||
`channel` VARCHAR(20) NOT NULL DEFAULT '',
|
||
`notified` TINYINT NOT NULL DEFAULT 0,
|
||
`created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||
KEY `idx_notified` (`notified`),
|
||
KEY `idx_task_created` (`task_name`, `created_at`)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='爬虫请求错误日志'""",
|
||
]
|
||
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
for ddl in ddls:
|
||
await cur.execute(ddl)
|
||
await self.ensure_alert_tables()
|
||
logger.info("数据库表检查/创建完成")
|
||
|
||
async def ensure_alert_tables(self):
|
||
await self.connect()
|
||
ddl = f"""CREATE TABLE IF NOT EXISTS `{self._prefix}crontab_alert` (
|
||
`id` BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
|
||
`crontab_id` INT NOT NULL DEFAULT 0,
|
||
`crontab_log_id` BIGINT UNSIGNED NOT NULL DEFAULT 0,
|
||
`task_name` VARCHAR(100) NOT NULL DEFAULT '',
|
||
`command` VARCHAR(100) NOT NULL DEFAULT '',
|
||
`params` VARCHAR(255) NOT NULL DEFAULT '',
|
||
`alert_type` VARCHAR(50) NOT NULL DEFAULT '',
|
||
`http_status` INT NOT NULL DEFAULT 0,
|
||
`detail` LONGTEXT NULL,
|
||
`dedupe_key` VARCHAR(191) NOT NULL DEFAULT '',
|
||
`status` VARCHAR(20) NOT NULL DEFAULT 'open',
|
||
`first_seen_at` INT UNSIGNED NOT NULL DEFAULT 0,
|
||
`last_seen_at` INT UNSIGNED NOT NULL DEFAULT 0,
|
||
`last_notified_at` INT UNSIGNED NOT NULL DEFAULT 0,
|
||
`resolved_at` INT UNSIGNED NOT NULL DEFAULT 0,
|
||
`notify_count` INT UNSIGNED NOT NULL DEFAULT 0,
|
||
`create_time` INT UNSIGNED NOT NULL DEFAULT 0,
|
||
`update_time` INT UNSIGNED NOT NULL DEFAULT 0,
|
||
UNIQUE KEY `uk_dedupe_key` (`dedupe_key`),
|
||
KEY `idx_status_notify` (`status`, `last_notified_at`),
|
||
KEY `idx_crontab_status` (`crontab_id`, `status`)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='定时任务告警'"""
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
await cur.execute(ddl)
|
||
|
||
# ── 彩种开奖结果 (la_lottery_draw_result) ──
|
||
|
||
async def upsert_lottery_draw_results(self, items: List[Dict[str, Any]]) -> int:
|
||
if not items:
|
||
return 0
|
||
await self.connect()
|
||
count = 0
|
||
now_ts = int(datetime.now().timestamp())
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
for item in items:
|
||
code = str(item.get('code', ''))
|
||
issue = str(item.get('issue', ''))
|
||
if not code or not issue:
|
||
continue
|
||
draw_time = item.get('draw_time') or None
|
||
draw_code = str(item.get('draw_code', ''))
|
||
draw_ext = str(item.get('draw_ext', ''))
|
||
status = int(item.get('status', 0))
|
||
if draw_code:
|
||
status = 1
|
||
idx = int(item.get('index', 0))
|
||
next_issue = str(item.get('next_issue', ''))
|
||
next_time = item.get('next_time') or None
|
||
trend = item.get('trend')
|
||
trend_json = json.dumps(trend, ensure_ascii=False) if trend else None
|
||
s_time = item.get('sTime') or None
|
||
await cur.execute(f"""
|
||
INSERT INTO `{self._prefix}lottery_draw_result`
|
||
(code, issue, draw_time, draw_code, draw_ext, status, idx,
|
||
next_issue, next_time, trend, s_time, create_time, update_time)
|
||
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
|
||
ON DUPLICATE KEY UPDATE
|
||
draw_time=VALUES(draw_time),
|
||
draw_code=VALUES(draw_code),
|
||
draw_ext=VALUES(draw_ext),
|
||
status=VALUES(status),
|
||
idx=VALUES(idx),
|
||
next_issue=VALUES(next_issue),
|
||
next_time=VALUES(next_time),
|
||
trend=VALUES(trend),
|
||
s_time=VALUES(s_time),
|
||
update_time=VALUES(update_time)
|
||
""", (
|
||
code, issue, draw_time, draw_code, draw_ext,
|
||
status, idx, next_issue, next_time, trend_json, s_time,
|
||
now_ts, now_ts,
|
||
))
|
||
count += 1
|
||
await conn.commit()
|
||
logger.info(f"la_lottery_draw_result UPSERT: {count} 条")
|
||
return count
|
||
|
||
# ── 超时比赛收尾 ──
|
||
|
||
async def get_overdue_matches(self, hours: int = 6, limit: int = 0) -> List[Dict]:
|
||
"""查询比赛时间超过 hours 小时但状态仍非已结束(status!=2)的比赛"""
|
||
await self.connect()
|
||
now_ts = int(datetime.now().timestamp())
|
||
cutoff = now_ts - hours * 3600
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor(aiomysql.DictCursor) as cur:
|
||
sql = f"""
|
||
SELECT id, match_id, home_team, away_team, league_name,
|
||
match_time, status, home_score, away_score, half_score
|
||
FROM `{self._prefix}match`
|
||
WHERE match_id > 0
|
||
AND match_time > 0
|
||
AND match_time <= %s
|
||
AND status IN (0, 1)
|
||
ORDER BY match_time ASC
|
||
"""
|
||
if limit > 0:
|
||
sql += f" LIMIT {limit}"
|
||
await cur.execute(sql, (cutoff,))
|
||
return await cur.fetchall()
|
||
|
||
async def finish_match(self, match_id: int, home_score: int, away_score: int, half_score: str = ""):
|
||
"""将比赛标记为已结束(status=2)，更新比分"""
|
||
await self.connect()
|
||
now_ts = int(datetime.now().timestamp())
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
await cur.execute(f"""
|
||
UPDATE `{self._prefix}match`
|
||
SET status=2, home_score=%s, away_score=%s,
|
||
half_score=IF(%s='', half_score, %s),
|
||
current_minute='FT', update_time=%s
|
||
WHERE match_id=%s AND status != 2
|
||
""", (home_score, away_score, half_score, half_score, now_ts, match_id))
|
||
|
||
async def has_match_lineup(self, match_id: int) -> bool:
|
||
"""检查比赛是否已有阵容数据"""
|
||
await self.connect()
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
await cur.execute(f"""
|
||
SELECT 1 FROM `{self._prefix}match_lineup`
|
||
WHERE match_id=%s LIMIT 1
|
||
""", (match_id,))
|
||
return await cur.fetchone() is not None
|
||
|
||
async def upsert_match_lineup(self, match_id: int, players: List[Dict[str, Any]]) -> int:
|
||
"""批量插入比赛阵容到 la_match_lineup（忽略重复）"""
|
||
if not players:
|
||
return 0
|
||
await self.connect()
|
||
now_ts = int(datetime.now().timestamp())
|
||
count = 0
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
for p in players:
|
||
try:
|
||
await cur.execute(f"""
|
||
INSERT IGNORE INTO `{self._prefix}match_lineup`
|
||
(match_id, team_side, is_starter, person_id, person_name,
|
||
person_logo, shirt_number, captain, position,
|
||
position_x, position_y, formation_place, create_time)
|
||
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
|
||
""", (
|
||
match_id,
|
||
int(p.get("team_side", 1)),
|
||
int(p.get("is_starter", 1)),
|
||
int(p.get("person_id", 0)),
|
||
str(p.get("person_name", ""))[:50],
|
||
str(p.get("person_logo", ""))[:500],
|
||
str(p.get("shirt_number", ""))[:10],
|
||
int(p.get("captain", 0)),
|
||
str(p.get("position", ""))[:20],
|
||
str(p.get("position_x", ""))[:10],
|
||
str(p.get("position_y", ""))[:10],
|
||
int(p.get("formation_place", 0)),
|
||
now_ts,
|
||
))
|
||
count += 1
|
||
except Exception:
|
||
pass
|
||
return count
|
||
|
||
@staticmethod
|
||
def _event_fingerprint(match_id: int, minute: str, etype: int, player: str, team_side: int) -> str:
|
||
raw = f"{match_id}:{minute}:{etype}:{player}:{team_side}"
|
||
return hashlib.md5(raw.encode()).hexdigest()
|
||
|
||
async def upsert_match_events(self, match_id: int, events: List[Dict[str, Any]]) -> int:
|
||
"""增量写入比赛事件，通过 Redis Set 缓存已有事件指纹，跳过重复"""
|
||
if not events:
|
||
return 0
|
||
await self.connect()
|
||
r = await self._get_redis()
|
||
now_ts = int(datetime.now().timestamp())
|
||
count = 0
|
||
EVENT_TYPE_MAP = {
|
||
"goal": 1, "assist": 2, "yellow": 3, "red": 4,
|
||
"substitution": 5, "penalty": 6, "own_goal": 7, "var": 8,
|
||
}
|
||
cache_key = f"match_event:{match_id}"
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
for ev in events:
|
||
minute = str(ev.get("minute", ""))[:10]
|
||
etype = EVENT_TYPE_MAP.get(str(ev.get("type", "")).lower(), 0)
|
||
team_side = int(ev.get("team_side", 1))
|
||
player = str(ev.get("player_name", "") or ev.get("player", ""))[:50]
|
||
desc = str(ev.get("description", "") or ev.get("text", ""))[:255]
|
||
fp = self._event_fingerprint(match_id, minute, etype, player, team_side)
|
||
if await r.sismember(cache_key, fp):
|
||
continue
|
||
try:
|
||
await cur.execute(f"""
|
||
INSERT IGNORE INTO `{self._prefix}match_event`
|
||
(match_id, minute, event_type, team_side, player_name, description, create_time)
|
||
VALUES (%s,%s,%s,%s,%s,%s,%s)
|
||
""", (match_id, minute, etype, team_side, player, desc, now_ts))
|
||
await r.sadd(cache_key, fp)
|
||
count += 1
|
||
except Exception:
|
||
pass
|
||
if count == 0:
|
||
await r.expire(cache_key, 86400)
|
||
else:
|
||
await r.expire(cache_key, 86400 * 3)
|
||
return count
|
||
|
||
# ── 爬虫错误日志 ──
|
||
|
||
async def insert_error_log(
|
||
self,
|
||
task_name: str,
|
||
request_url: str,
|
||
request_method: str = "GET",
|
||
request_params: str = "",
|
||
http_status: int = 0,
|
||
error_type: str = "",
|
||
error_message: str = "",
|
||
response_body: str = "",
|
||
channel: str = "",
|
||
):
|
||
await self.connect()
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
await cur.execute(f"""
|
||
INSERT INTO `{self._prefix}crawl_error_log`
|
||
(task_name, request_url, request_method, request_params,
|
||
http_status, error_type, error_message, response_body, channel)
|
||
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)
|
||
""", (
|
||
task_name[:100], request_url[:1000], request_method[:10],
|
||
request_params[:5000] if request_params else "",
|
||
http_status, error_type[:100],
|
||
error_message[:5000] if error_message else "",
|
||
response_body[:5000] if response_body else "",
|
||
channel[:20],
|
||
))
|
||
|
||
async def get_pending_errors(self, limit: int = 100) -> List[Dict]:
|
||
await self.connect()
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor(aiomysql.DictCursor) as cur:
|
||
await cur.execute(f"""
|
||
SELECT id, task_name, request_url, request_method, request_params,
|
||
http_status, error_type, error_message, response_body,
|
||
channel, created_at
|
||
FROM `{self._prefix}crawl_error_log`
|
||
WHERE notified = 0
|
||
ORDER BY created_at ASC
|
||
LIMIT %s
|
||
""", (limit,))
|
||
return await cur.fetchall()
|
||
|
||
async def mark_errors_notified(self, ids: List[int]):
|
||
if not ids:
|
||
return
|
||
await self.connect()
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
placeholders = ','.join(['%s'] * len(ids))
|
||
await cur.execute(f"""
|
||
UPDATE `{self._prefix}crawl_error_log`
|
||
SET notified = 1
|
||
WHERE id IN ({placeholders})
|
||
""", ids)
|
||
|
||
async def upsert_crontab_alert(
|
||
self,
|
||
crontab_id: int,
|
||
crontab_log_id: int,
|
||
task_name: str,
|
||
command: str,
|
||
params: str,
|
||
alert_type: str,
|
||
http_status: int,
|
||
detail: Dict[str, Any],
|
||
dedupe_key: str,
|
||
) -> int:
|
||
await self.ensure_alert_tables()
|
||
now_ts = int(datetime.now().timestamp())
|
||
detail_json = json.dumps(detail, ensure_ascii=False) if isinstance(detail, (dict, list)) else str(detail or "")
|
||
existing = await self.fetchone(
|
||
f"SELECT id, status FROM `{self._prefix}crontab_alert` WHERE dedupe_key=%s LIMIT 1",
|
||
(dedupe_key,),
|
||
)
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
if existing:
|
||
if existing.get("status") == "resolved":
|
||
await cur.execute(
|
||
f"""UPDATE `{self._prefix}crontab_alert`
|
||
SET crontab_id=%s, crontab_log_id=%s, task_name=%s, `command`=%s, params=%s,
|
||
alert_type=%s, http_status=%s, detail=%s,
|
||
status='open', first_seen_at=%s, last_seen_at=%s,
|
||
last_notified_at=0, resolved_at=0, notify_count=0, update_time=%s
|
||
WHERE id=%s""",
|
||
(
|
||
crontab_id, crontab_log_id, task_name[:100], command[:100], params[:255],
|
||
alert_type[:50], int(http_status), detail_json,
|
||
now_ts, now_ts, now_ts, existing["id"],
|
||
),
|
||
)
|
||
else:
|
||
await cur.execute(
|
||
f"""UPDATE `{self._prefix}crontab_alert`
|
||
SET crontab_log_id=%s, task_name=%s, `command`=%s, params=%s,
|
||
alert_type=%s, http_status=%s, detail=%s,
|
||
last_seen_at=%s, update_time=%s
|
||
WHERE id=%s""",
|
||
(
|
||
crontab_log_id, task_name[:100], command[:100], params[:255],
|
||
alert_type[:50], int(http_status), detail_json,
|
||
now_ts, now_ts, existing["id"],
|
||
),
|
||
)
|
||
return int(existing["id"])
|
||
|
||
await cur.execute(
|
||
f"""INSERT INTO `{self._prefix}crontab_alert`
|
||
(crontab_id, crontab_log_id, task_name, `command`, params, alert_type, http_status,
|
||
detail, dedupe_key, status, first_seen_at, last_seen_at, create_time, update_time)
|
||
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,'open',%s,%s,%s,%s)""",
|
||
(
|
||
crontab_id, crontab_log_id, task_name[:100], command[:100], params[:255],
|
||
alert_type[:50], int(http_status), detail_json, dedupe_key[:191],
|
||
now_ts, now_ts, now_ts, now_ts,
|
||
),
|
||
)
|
||
return int(cur.lastrowid)
|
||
|
||
async def resolve_crontab_alerts(self, crontab_id: int):
|
||
await self.ensure_alert_tables()
|
||
now_ts = int(datetime.now().timestamp())
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
await cur.execute(
|
||
f"""UPDATE `{self._prefix}crontab_alert`
|
||
SET status='resolved', resolved_at=%s, update_time=%s
|
||
WHERE crontab_id=%s AND status='open'""",
|
||
(now_ts, now_ts, crontab_id),
|
||
)
|
||
|
||
async def get_dispatchable_crontab_alerts(self, cooldown_seconds: int, limit: int = 50) -> List[Dict]:
|
||
await self.ensure_alert_tables()
|
||
now_ts = int(datetime.now().timestamp())
|
||
cutoff = max(0, now_ts - max(0, int(cooldown_seconds)))
|
||
return await self.fetchall(
|
||
f"""SELECT a.*, c.expression
|
||
FROM `{self._prefix}crontab_alert` a
|
||
LEFT JOIN `{self._prefix}dev_crontab` c ON c.id = a.crontab_id
|
||
WHERE a.status='open'
|
||
AND (a.last_notified_at = 0 OR a.last_notified_at <= %s)
|
||
ORDER BY a.first_seen_at ASC
|
||
LIMIT %s""",
|
||
(cutoff, int(limit)),
|
||
)
|
||
|
||
async def mark_crontab_alert_notified(self, alert_id: int):
|
||
await self.ensure_alert_tables()
|
||
now_ts = int(datetime.now().timestamp())
|
||
async with self._pool.acquire() as conn:
|
||
async with conn.cursor() as cur:
|
||
await cur.execute(
|
||
f"""UPDATE `{self._prefix}crontab_alert`
|
||
SET last_notified_at=%s, notify_count=notify_count + 1, update_time=%s
|
||
WHERE id=%s""",
|
||
(now_ts, now_ts, int(alert_id)),
|
||
)
|
||
|
||
async def __aenter__(self):
|
||
await self.connect()
|
||
return self
|
||
|
||
async def __aexit__(self, *args):
|
||
await self.close()
|