import sqlite3
|
import os
|
from collections import defaultdict
|
from datetime import datetime
|
from auto_media_publisher.config.conf_base import DATABASE_PATH
|
from auto_media_publisher.utils.logger import get_logger
|
|
|
DATABASE_FLAG_FILE = DATABASE_PATH.replace(".sqlite3",".initialized")
|
|
database_logger = get_logger("database","admin")
|
|
def _init_database():
|
if os.path.exists(DATABASE_FLAG_FILE):
|
return
|
|
with sqlite3.connect(DATABASE_PATH) as conn:
|
cursor = conn.cursor()
|
cursor.execute('''
|
CREATE TABLE IF NOT EXISTS publish_tasks (
|
"id" INTEGER PRIMARY KEY AUTOINCREMENT,
|
"user_id" TEXT,
|
"platform" TEXT,
|
"generator_class" TEXT,
|
"generator_args" TEXT,
|
"scheduled_time" DATETIME,
|
"created_time" DATETIME,
|
"media_time" DATETIME,
|
"upload_time" DATETIME,
|
"task_status" TEXT DEFAULT 'pending',
|
"config_updated" TEXT,
|
"proxy_full" TEXT,
|
"note" TEXT
|
)
|
''')
|
cursor.execute('''
|
CREATE TABLE IF NOT EXISTS dajiala_top15_urls (
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
url TEXT UNIQUE,
|
title TEXT,
|
author TEXT,
|
content TEXT,
|
type INTEGER,
|
type_name TEXT,
|
updatetime TEXT,
|
status TEXT DEFAULT 'pending',
|
retry_count INTEGER DEFAULT 0,
|
last_update TEXT
|
)
|
''')
|
cursor.execute('''
|
CREATE TABLE IF NOT EXISTS operation_fails (
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
user TEXT,
|
platform TEXT,
|
proxy TEXT,
|
type_ TEXT, --proxy --platform
|
created_time DATETIME,
|
note TEXT
|
)
|
''')
|
conn.commit()
|
|
# 创建标志文件
|
with open(DATABASE_FLAG_FILE, 'w') as f:
|
f.write("Database initialized")
|
database_logger.info("Database initialized!!!")
|
|
def dajiala_insert_or_update_top15_json(json_obj):
|
now = datetime.now().isoformat()
|
|
# 提取字段（字段名需与你表结构一致）
|
url = json_obj.get("url")
|
title = json_obj.get("title", "")
|
author = json_obj.get("wx_name", "") # 对应字段改名
|
content = json_obj.get("content", "") # 若无内容字段，可为空
|
type_val = json_obj.get("type", 0)
|
type_name = json_obj.get("type_name", "未知")
|
updatetime = json_obj.get("updatetime", now)
|
|
with sqlite3.connect(DATABASE_PATH) as conn:
|
cursor = conn.cursor()
|
cursor.execute('''
|
INSERT INTO dajiala_top15_urls
|
(url, title, author, content, type, type_name, updatetime, status, retry_count, last_update)
|
VALUES (?, ?, ?, ?, ?, ?, ?, 'pending', 0, ?)
|
ON CONFLICT(url) DO UPDATE SET
|
title=excluded.title,
|
author=excluded.author,
|
content=excluded.content,
|
type=excluded.type,
|
type_name=excluded.type_name,
|
updatetime=excluded.updatetime,
|
last_update=excluded.last_update
|
''', (
|
url, title, author, content,
|
type_val, type_name, updatetime, now
|
))
|
conn.commit()
|
|
def dajiala_insert_or_update_top15(url,title,author,content,type_val,type_name,updatetime):
|
now = datetime.now().isoformat()
|
with sqlite3.connect(DATABASE_PATH) as conn:
|
cursor = conn.cursor()
|
cursor.execute('''
|
INSERT INTO dajiala_top15_urls
|
(url, title, author, content, type, type_name, updatetime, status, retry_count, last_update)
|
VALUES (?, ?, ?, ?, ?, ?, ?, 'pending', 0, ?)
|
ON CONFLICT(url) DO UPDATE SET
|
title=excluded.title,
|
author=excluded.author,
|
content=excluded.content,
|
type=excluded.type,
|
type_name=excluded.type_name,
|
updatetime=excluded.updatetime,
|
last_update=excluded.last_update
|
''', (
|
url, title, author, content,
|
type_val, type_name, updatetime, now
|
))
|
conn.commit()
|
|
def dajiala_get_pending_downloads(limit=200):
|
with sqlite3.connect(DATABASE_PATH) as conn:
|
conn.row_factory = sqlite3.Row # ✅ 支持字段名访问
|
cursor = conn.cursor()
|
cursor.execute('''
|
SELECT * FROM dajiala_top15_urls
|
WHERE status != 'done'
|
ORDER BY last_update ASC
|
LIMIT ?
|
''', (limit,))
|
return cursor.fetchall()
|
|
def dajiala_update_download_status(url, status):
|
with sqlite3.connect(DATABASE_PATH) as conn:
|
cursor = conn.cursor()
|
if status == 'done':
|
cursor.execute('''
|
UPDATE dajiala_top15_urls SET status = ?, retry_count = retry_count, last_update = ?
|
WHERE url = ?
|
''', (status, datetime.now().isoformat(), url))
|
else:
|
cursor.execute('''
|
UPDATE dajiala_top15_urls SET retry_count = retry_count + 1, last_update = ?
|
WHERE url = ?
|
''', (datetime.now().isoformat(), url))
|
conn.commit()
|
|
def media_get_tasks(limit=200):
|
with sqlite3.connect(DATABASE_PATH) as conn:
|
conn.row_factory = sqlite3.Row # ✅ 支持字段名访问
|
cursor = conn.cursor()
|
cursor.execute('''
|
SELECT * FROM publish_tasks
|
WHERE task_status = 'pending' and scheduled_time > datetime('now', '+8 hours')
|
ORDER BY created_time ASC
|
LIMIT ?
|
''', (limit,))
|
rows = cursor.fetchall()
|
return rows # 每个 row 支持 row["id"]、row["created_time"] 等
|
|
def media_update_task_done(task_id, media_time):
|
try:
|
with sqlite3.connect(DATABASE_PATH) as conn:
|
conn.execute("""
|
UPDATE publish_tasks set media_time = ? , task_status = 'media_done'
|
WHERE id = ?
|
""", (media_time,task_id))
|
conn.commit()
|
return True
|
except Exception as e:
|
database_logger.error("Database error!!! {e}")
|
return False
|
|
def publish_task_exists(user_id, platform, scheduled_time):
|
with sqlite3.connect(DATABASE_PATH) as conn:
|
conn.row_factory = sqlite3.Row # ✅ 支持字段名访问
|
cursor = conn.cursor()
|
cursor.execute("""
|
SELECT id, config_updated, task_status FROM publish_tasks
|
WHERE user_id=? AND platform=? AND scheduled_time=?
|
""", (user_id, platform, scheduled_time))
|
return cursor.fetchone()
|
|
def publish_task_insert(user_id, platform, generator_class,generator_args,scheduled_time, config_updated,proxy_full):
|
try:
|
with sqlite3.connect(DATABASE_PATH) as conn:
|
conn.execute("""
|
INSERT INTO publish_tasks (user_id, platform, generator_class,generator_args,scheduled_time,created_time,config_updated,proxy_full)
|
VALUES (?, ?, ?, ?, ?,?, ?,?)
|
""", (user_id, platform,generator_class,generator_args,scheduled_time,datetime.now().strftime("%Y-%m-%d %H:%M:%S"),config_updated,proxy_full))
|
conn.commit()
|
return True
|
except Exception as e:
|
database_logger.error("Database error!!! {e}")
|
return False
|
|
def publish_task_delete(task_id):
|
try:
|
with sqlite3.connect(DATABASE_PATH) as conn:
|
conn.execute("DELETE FROM publish_tasks WHERE id=?", (task_id,))
|
conn.commit()
|
return True
|
except Exception as e:
|
database_logger.error("Database error!!! {e}")
|
return False
|
|
def platform_operation_fail_insert(user,platform,proxy,type_,note):
|
try:
|
with sqlite3.connect(DATABASE_PATH) as conn:
|
conn.execute("""
|
INSERT INTO operation_fails (user, platform, proxy, type_,note)
|
VALUES (?, ?, ?, ?, ?)
|
""", (user, platform,proxy,type_,note))
|
conn.commit()
|
return True
|
except Exception as e:
|
database_logger.error("Database error!!! {e}")
|
return False
|
|
def get_platform_upload_tasks(limit=200):
|
with sqlite3.connect(DATABASE_PATH) as conn:
|
conn.row_factory = sqlite3.Row # ✅ 支持字段名访问
|
cursor = conn.cursor()
|
cursor.execute('''
|
SELECT id,user_id as user,platform,proxy_full as proxy, strftime('%Y%m%d%H%M%S', scheduled_time) as scheduled_time FROM publish_tasks
|
WHERE task_status = 'media_done' and scheduled_time > datetime('now', '+8 hours')
|
ORDER BY created_time ASC
|
LIMIT ?
|
''', (limit,))
|
rows = cursor.fetchall()
|
grouped = defaultdict(list)
|
for task in rows:
|
key = (task['user'], task['platform'], task['proxy'])
|
grouped[key].append({'id': task['id'], 'scheduled_time': task['scheduled_time']})
|
|
# 生成目标结构
|
result = []
|
for (user, platform, proxy), scheduled_times in grouped.items():
|
result.append({
|
'user': user,
|
'platform': platform,
|
'proxy': proxy,
|
'scheduled_times': scheduled_times
|
})
|
|
return result
|
|
def update_task_upload_done(task_id, upload_time,note):
|
try:
|
with sqlite3.connect(DATABASE_PATH) as conn:
|
conn.execute("""
|
UPDATE publish_tasks set upload_time = ? , task_status = 'done' , note = ?
|
WHERE id = ?
|
""", (upload_time,note,task_id))
|
conn.commit()
|
return True
|
except Exception as e:
|
database_logger.error("Database error!!! {e}")
|
return False
|
