192 lines
6.8 KiB
Python
Executable File
192 lines
6.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
import os
|
||
import re
|
||
import sys
|
||
import time
|
||
import logging
|
||
import requests
|
||
import sqlite3
|
||
from requests.auth import HTTPBasicAuth
|
||
from requests.exceptions import RequestException
|
||
|
||
# Paperless 服务器信息
|
||
PAPERLESS_URL = "http://localhost:8000/api"
|
||
#AUTH = HTTPBasicAuth("admin", "admin") # Basic Auth 认证， mac上用这个
|
||
AUTH = HTTPBasicAuth("admin", "paperless") # Basic Auth 认证，NAS上用这个
|
||
|
||
# 日志配置
|
||
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
|
||
|
||
|
||
# 连接到 SQLite 数据库
|
||
#DB_PATH = "/usr/src/paperless/db/paperless.sqlite3" # 根据你的数据库实际路径调整
|
||
DB_PATH = "/usr/src/paperless/data/db.sqlite3"
|
||
conn = sqlite3.connect(DB_PATH)
|
||
cursor = conn.cursor()
|
||
enable_db = False # 标准用法，用API
|
||
|
||
# 正则解析文件名
|
||
FILENAME_PATTERN = re.compile(r"(\d{4}-\d{2}-\d{2})_(.*?)_(.*?)_(.*?)_(.*?)_(.*)\.pdf")
|
||
|
||
# 直接从db里获取数据
|
||
def query_id_from_db(table, column, value):
|
||
try:
|
||
cursor.execute(f"SELECT id FROM {table} WHERE {column} = ?", (value,))
|
||
row = cursor.fetchone()
|
||
if row:
|
||
logging.info(f'query {table} where {column}="{value}" get id={row[0]}')
|
||
return row[0]
|
||
else:
|
||
return None
|
||
except sqlite3.Error as e:
|
||
logging.error(f"Error inserting or updating data: {e}")
|
||
return None
|
||
|
||
# API 请求封装（带重试）,支持GET，POST等
|
||
def api_request(method, url, data=None, retries=5):
|
||
for attempt in range(retries):
|
||
try:
|
||
response = requests.request(method, url, json=data, auth=AUTH, timeout=5)
|
||
|
||
if response.status_code in [200, 201]:
|
||
return response.json()
|
||
elif response.status_code == 404:
|
||
logging.warning(f"API 资源未找到: {method} {url}")
|
||
return None
|
||
else:
|
||
logging.error(f"API 请求失败: {method} {url}, 状态码: {response.status_code}, 响应: {response.text}")
|
||
|
||
except RequestException as e:
|
||
logging.error(f"API 请求异常: {method} {url}, 错误: {e}")
|
||
|
||
if attempt < retries - 1:
|
||
logging.warning(f"请求失败，等待 2 秒后重试 ({attempt+1}/{retries})...")
|
||
time.sleep(2)
|
||
|
||
logging.error(f"API 请求最终失败: {method} {url}")
|
||
return None
|
||
|
||
# 从API查询
|
||
def query_id_from_rest_api(endpoint, name):
|
||
url = f"{PAPERLESS_URL}/{endpoint}"
|
||
method = 'GET'
|
||
while url is not None : # 循环请求所有分页数据
|
||
json_data = api_request(method, url, data=None, retries=5)
|
||
if json_data:
|
||
if "results" in json_data and isinstance(json_data["results"], list):
|
||
# 找到就直接返回了
|
||
for item in json_data.get('results', []):
|
||
if item["name"] == name:
|
||
return item["id"]
|
||
# 没找到，继续下一页
|
||
url = json_data.get("next", None)
|
||
else:
|
||
logging.warning(f"API 返回无数据: {method} {url}")
|
||
break
|
||
else:
|
||
logging.warning(f"API 返回无数据: {method} {url}")
|
||
break
|
||
return None
|
||
|
||
# 获取或创建文档类型
|
||
def get_or_create_document_type(name):
|
||
if enable_db:
|
||
id = query_id_from_db("documents_documenttype", "name", name)
|
||
else:
|
||
id = query_id_from_rest_api("document_types/", name)
|
||
if id:
|
||
return id
|
||
|
||
# 创建新的文档类型
|
||
new_doc_type = api_request("POST", f"{PAPERLESS_URL}/document_types/", {"name": name})
|
||
return new_doc_type.get("id") if new_doc_type else None
|
||
|
||
# 获取或创建 Correspondent（机构/通信者）
|
||
def get_or_create_correspondent(name):
|
||
if enable_db:
|
||
id = query_id_from_db("documents_correspondent", "name", name)
|
||
else:
|
||
id = query_id_from_rest_api("correspondents/", name)
|
||
if id:
|
||
return id
|
||
|
||
# 创建新的 Correspondent
|
||
new_correspondent = api_request("POST", f"{PAPERLESS_URL}/correspondents/", {"name": name})
|
||
return new_correspondent.get("id") if new_correspondent else None
|
||
|
||
|
||
# 获取或创建自定义字段
|
||
def get_or_create_custom_field(name):
|
||
if enable_db:
|
||
id = query_id_from_db("documents_customfield", "name", name)
|
||
else:
|
||
id = query_id_from_rest_api("custom_fields/", name)
|
||
if id:
|
||
return id
|
||
|
||
# 创建新的自定义字段
|
||
new_field = api_request("POST", f"{PAPERLESS_URL}/custom_fields/", {"name": name, "data_type": "string"})
|
||
return new_field.get("id") if new_field else None
|
||
|
||
# 更新文档信息
|
||
def update_document(document_id, title, publish_date, doc_type_id, correspondent_id, custom_fields):
|
||
payload = {
|
||
"title": title,
|
||
"created_date": publish_date,
|
||
"document_type": doc_type_id,
|
||
"correspondent": correspondent_id,
|
||
"custom_fields": custom_fields,
|
||
}
|
||
return api_request("PATCH", f"{PAPERLESS_URL}/documents/{document_id}/", payload) is not None
|
||
|
||
# 主函数 - 解析文件名并更新 Paperless
|
||
def process_document():
|
||
doc_id = os.getenv("DOCUMENT_ID")
|
||
filename = os.getenv("DOCUMENT_ORIGINAL_FILENAME")
|
||
|
||
# debug
|
||
#doc_id = sys.argv[1]
|
||
#filename = sys.argv[2]
|
||
|
||
if not doc_id or not filename:
|
||
logging.error("❌ 缺少必要的环境变量，无法执行脚本")
|
||
return
|
||
|
||
match = FILENAME_PATTERN.match(filename)
|
||
if not match:
|
||
logging.warning(f"⚠️ 文件名格式不匹配，跳过处理: {filename}")
|
||
return
|
||
|
||
publish_date, report_type, org_sname, industry_name, stock_name, title = match.groups()
|
||
|
||
logging.info(f"✅ 解析成功: 日期={publish_date}, 类型={report_type}, 机构={org_sname}, 行业={industry_name}, 股票={stock_name}, 标题={title}")
|
||
|
||
# 获取或创建文档类型 & Correspondent
|
||
doc_type_id = get_or_create_document_type(report_type)
|
||
correspondent_id = get_or_create_correspondent(org_sname)
|
||
if not doc_type_id or not correspondent_id:
|
||
logging.error("❌ 文档类型/机构创建失败")
|
||
return
|
||
|
||
# 获取或创建自定义字段
|
||
industry_field_id = get_or_create_custom_field("行业")
|
||
stock_field_id = get_or_create_custom_field("股票名称")
|
||
if not industry_field_id or not stock_field_id:
|
||
logging.error("❌ 自定义字段创建失败")
|
||
return
|
||
|
||
# 组装自定义字段数据
|
||
custom_fields = [
|
||
{"field": industry_field_id, "value": industry_name},
|
||
{"field": stock_field_id, "value": stock_name}
|
||
]
|
||
|
||
# 更新 Paperless 文档
|
||
success = update_document(doc_id, title, publish_date, doc_type_id, correspondent_id, custom_fields)
|
||
if success:
|
||
logging.info(f"✅ 文档 {doc_id} 更新成功: {title}")
|
||
else:
|
||
logging.error(f"❌ 文档 {doc_id} 更新失败")
|
||
|
||
if __name__ == "__main__":
|
||
process_document() |