用 Paperless-ngx 整理十年文档：从 Google Drive 文件夹到全文检索归档库

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

10 - 文书材料/
  10 - 证件材料/身份证件/
  30 - 移民文档/
  30 - Tax Filing/
  40 - Finance/
  50 - 车辆注册/
  60 - 住房买房/
  80 - Medical/
20 - 家装住房信息/
80 - 旅行计划/

1
2
3
4
5
6
7
8
9

Google Takeout .zip
  │  migrate.py（分类 + 上传）
  ▼
Paperless-ngx（Docker，10.0.10.10:5656）
  │  REST API  POST /api/documents/post_document/
  │  OCR + 全文索引
  │  ML 分类器（在已标注语料上重训练）
  ▼
NAS 存储（/volume1/docker/paperless/）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# 关闭所有文档类型的匹配
curl -s "http://10.0.10.10:5656/api/document_types/?page_size=50" \
  -H "Authorization: Token YOUR_TOKEN" | jq '.results[] | .id' | \
while read id; do
  curl -s -X PATCH "http://10.0.10.10:5656/api/document_types/$id/" \
    -H "Authorization: Token YOUR_TOKEN" \
    -H "Content-Type: application/json" \
    -d '{"matching_algorithm": 0}'
done

# 标签和往来方同理
# matching_algorithm: 0=无, 1=任意, 2=全部, 3=字面, 4=正则, 6=Auto/ML

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

def classify(rel_path: str) -> dict:
    p = rel_path
    filename = Path(p).name
    year = extract_year(Path(p).parts)  # 从路径任意部分提取年份
    corr = corr_from_filename(filename) # 从文件名关键词匹配往来方

    if "10 - 身份证件" in p:
        return result(DOCTYPE["身份证件"], [TAG["移民"]])

    if "30 - 移民文档" in p:
        return result(DOCTYPE["移民文件"], [TAG["移民"]])

    if "30 - Tax Filing" in p:
        tax_tags = [TAG["税务"]]
        if re.search(r'w[\-\s]?2\b', fl):
            return result(DOCTYPE["税务文件"], tax_tags, corr or CORR["Google LLC"])
        if re.search(r'(f?1099|f?1098|f?5498|1095)', fl):
            return result(DOCTYPE["税务文件"], tax_tags, corr)
        return result(DOCTYPE["税务文件"], tax_tags)

    # ... 更多规则 ...

    return result(None, [TAG["待处理"]])  # 兜底：进收件箱

1
2
3
4
5
6
7

resp = requests.post(
    f"{API_BASE}/documents/post_document/",
    headers={"Authorization": f"Token {API_TOKEN}"},
    files={"document": (filename, data)},
    data=form_data,   # (key, value) 元组列表，支持重复字段
    timeout=120,
)

1
2

for tag_id in meta["tags"]:
    form_data.append(("tags", tag_id))

1

form_data.append(("custom_fields", json.dumps({str(YEAR_FIELD_ID): year})))

1
2
3
4
5
6

SKIP_EXTENSIONS = {
    ".html", ".csv", ".qfx", ".gsheet", ".gdoc", ".gslides",
    ".gdraw", ".gmap", ".java", ".bin", ".db", ".exe", ".7z",
    ".rar", ".gshortcut", ".pptx", ".xls", ".xlsx", ".tar",
    ".doc", ".docx",  # Paperless OCR 支持不稳定
}

1

python3 migrate.py --dry-run

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

=================================================================
  IMPORT PREVIEW — 358 files to import
=================================================================

By document type:
  税务文件        89 files
  移民文件        72 files
  金融账单        61 files
  ...

Skipped (non-document files): 47

Sample mappings (first 30):
  10 - 文书材料/30 - Tax Filing/2022/W2-Google.pdf
    → 税务文件 | tags:['税务'] | corr:Google LLC | year:2022
    title: W2-Google

1

python3 migrate.py --execute --yes

1
2
3

# 在 NAS 上运行（需要 docker 权限）
/usr/local/bin/docker exec paperless-webserver-1 \
  python3 manage.py document_create_classifier

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

import requests

API_BASE  = "http://10.0.10.10:5656/api"
API_TOKEN = "YOUR_TOKEN"
TAG_ID    = 7  # #待处理

headers = {"Authorization": f"Token {API_TOKEN}",
           "Content-Type": "application/json"}

r = requests.get(f"{API_BASE}/documents/?tags__id__all={TAG_ID}&page_size=500",
                 headers=headers)
doc_ids = [d["id"] for d in r.json()["results"]]
print(f"待清理文档数：{len(doc_ids)}")

r = requests.post(f"{API_BASE}/documents/bulk_edit/",
    headers=headers,
    json={"documents": doc_ids,
          "method": "remove_tag",
          "parameters": {"tag": TAG_ID}})
print(r.status_code, r.text)

1
2

/usr/local/bin/docker exec paperless-webserver-1 \
  python3 manage.py document_create_classifier

1
2
3
4
5

# ✓ 正确
json.dumps({"4": 2024})

# ✗ 错误——返回 400 错误
json.dumps([{"field": 4, "value": 2024}])