kali/DailyMotion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: operatetime 时间去除

Browse Source
This commit is contained in:
晓丰 2025-07-08 19:44:47 +08:00
parent 2a08c78b89
commit 2c416d8543
1 changed files with 8 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
DB.py
View File

@ -704,10 +704,6 @@ class DBSA:
# ---------------------------------------------------- # ----------------------------------------------------
@classmethod @classmethod
def upsert_video(cls, data: dict): def upsert_video(cls, data: dict):
"""
业务线程/进程调用此方法写入。
如果启用了 queue 模式，则把 data 丢队列即可。
"""
if cls._queue_mode: if cls._queue_mode:
cls._queue.put(data) cls._queue.put(data)
return return
@ -721,7 +717,6 @@ class DBSA:
now_ts = int(time.time()) now_ts = int(time.time())
op_index_key = (data["v_xid"] or "", data["keyword"] or "", now_ts) op_index_key = (data["v_xid"] or "", data["keyword"] or "", now_ts)
# —— 用 v_name 去重，避免 title 冲突 ——
vid_index_key = (data["v_xid"] or "", data["v_name"] or "") vid_index_key = (data["v_xid"] or "", data["v_name"] or "")
# ---------- ① 获取互斥锁 ---------- # ---------- ① 获取互斥锁 ----------
@ -739,13 +734,12 @@ class DBSA:
v_xid=data["v_xid"], v_xid=data["v_xid"],
a_id=data["a_id"], a_id=data["a_id"],
level=data.get("level", 0), level=data.get("level", 0),
name_title=data["v_name"], # ✨ NEW: 截断避免过长 name_title=data["v_name"],
keyword=data["keyword"], keyword=data["keyword"],
is_repeat=data["is_repeat"], is_repeat=data["is_repeat"],
sort=data["sort"], sort=data["sort"],
createtime=now_ts, createtime=now_ts, # 首次插入
updatetime=now_ts, updatetime=now_ts, # 后续更新只改这一列
operatetime=now_ts,
batch=data.get("batch", 0), batch=data.get("batch", 0),
machine=data.get("machine_id", 0), machine=data.get("machine_id", 0),
is_piracy=data.get("is_piracy", '3'), is_piracy=data.get("is_piracy", '3'),
@ -772,7 +766,6 @@ class DBSA:
status=1, status=1,
createtime=now_ts, createtime=now_ts,
updatetime=now_ts, updatetime=now_ts,
operatetime=now_ts,
watch_number=data.get("view", 0), watch_number=data.get("view", 0),
follow_number=data.get("fans", 0), follow_number=data.get("fans", 0),
video_number=data.get("videos", 0), video_number=data.get("videos", 0),
@ -891,10 +884,9 @@ class DBSA:
if op_rows: if op_rows:
stmt_op = mysql_insert(video_op).values(op_rows) stmt_op = mysql_insert(video_op).values(op_rows)
ondup_op = stmt_op.on_duplicate_key_update( ondup_op = stmt_op.on_duplicate_key_update(
updatetime = stmt_op.inserted.updatetime, updatetime = stmt_op.inserted.updatetime,
operatetime = stmt_op.inserted.operatetime, ts_status = stmt_op.inserted.ts_status,
ts_status = stmt_op.inserted.ts_status, is_repeat = stmt_op.inserted.is_repeat,
is_repeat = stmt_op.inserted.is_repeat,
) )
cls._safe_execute(ondup_op, desc="video_op") cls._safe_execute(ondup_op, desc="video_op")
logger.info("落表：操作记录 %d 条", len(op_rows)) logger.info("落表：操作记录 %d 条", len(op_rows))
@ -920,7 +912,6 @@ class DBSA:
u_name = stmt_vid.inserted.u_name, u_name = stmt_vid.inserted.u_name,
status = stmt_vid.inserted.status, status = stmt_vid.inserted.status,
updatetime = stmt_vid.inserted.updatetime, updatetime = stmt_vid.inserted.updatetime,
operatetime = stmt_vid.inserted.operatetime,
) )
cls._safe_execute(ondup_vid, desc="video") cls._safe_execute(ondup_vid, desc="video")
logger.info("落表：视频记录 %d 条", len(vid_rows)) logger.info("落表：视频记录 %d 条", len(vid_rows))
@ -944,9 +935,6 @@ class DBSA:
# ---------------------------------------------------- # ----------------------------------------------------
@classmethod @classmethod
def start_single_flusher(cls): def start_single_flusher(cls):
"""
启动后台线程 —— 生产线程只喂 queue，flusher 串行写库，彻底避免锁竞争。
"""
cls._queue_mode = True cls._queue_mode = True
def _worker(): def _worker():
@ -955,19 +943,17 @@ class DBSA:
try: try:
data = cls._queue.get(timeout=3) data = cls._queue.get(timeout=3)
batch.append(data) batch.append(data)
# drain 队列
while True: while True:
try: try:
batch.append(cls._queue.get_nowait()) batch.append(cls._queue.get_nowait())
except Empty: except Empty:
break break
except Empty: except Empty:
pass # 队列暂时为空 pass
if not batch: if not batch:
continue continue
# ---- 把 batch 数据重新写入缓冲（无锁）----
for d in batch: for d in batch:
cls._buffer_without_lock(d) cls._buffer_without_lock(d)
batch.clear() batch.clear()
@ -994,7 +980,6 @@ class DBSA:
if op_key in cls._existing_op_keys or vid_key in cls._existing_vid_keys: if op_key in cls._existing_op_keys or vid_key in cls._existing_vid_keys:
return return
# —— op_row 同构 ——
op_row = dict( op_row = dict(
v_id=data["v_id"], v_id=data["v_id"],
v_xid=data["v_xid"], v_xid=data["v_xid"],
@ -1006,7 +991,6 @@ class DBSA:
sort=data["sort"], sort=data["sort"],
createtime=now_ts, createtime=now_ts,
updatetime=now_ts, updatetime=now_ts,
operatetime=now_ts,
batch=data.get("batch", 0), batch=data.get("batch", 0),
machine=data.get("machine_id", 0), machine=data.get("machine_id", 0),
is_piracy=data.get("is_piracy", '3'), is_piracy=data.get("is_piracy", '3'),
@ -1031,7 +1015,6 @@ class DBSA:
status=1, status=1,
createtime=now_ts, createtime=now_ts,
updatetime=now_ts, updatetime=now_ts,
operatetime=now_ts,
watch_number=data.get("view", 0), watch_number=data.get("view", 0),
follow_number=data.get("fans", 0), follow_number=data.get("fans", 0),
video_number=data.get("videos", 0), video_number=data.get("videos", 0),