django-log-archive

日志归档通用扩展包：抽象归档模型、分块归档流水线、可扩展统计聚合、流式归档加解密。

特性

• 抽象归档模型 — ArchiveRecordBase / ArchiveStatsBase，业务继承即得归档记录与按日统计表，无实质 DDL
• 分块归档流水线 — ChunkedPipeline / PipelineStage / iter_batches，大表按批次取数、写归档、统计、删记录；内置通用阶段 ArchiveExportStage（按日追加式导出）/ SourceCleanupStage（归档后删源数据）与集成默认阶段的 ArchivePipeline 基类
• 原始日志保留策略 — ArchivePipeline(retention_days=N) 实现「按日归档、原始日志保存 N 天」：归档时打 archived_at 标记而非立即删除，到期后自动清理；重复归档不会产生重复数据（DJANGO_LOG_ARCHIVE_RETENTION_DAYS 可配默认值）
• 可扩展统计聚合 — BaseAggregate + 内置 CountAggregate / SuccessFailureAggregate / TokensAggregate，StatsAggregator 驱动分组聚合
• 流式归档加解密 — 64KB 分块 + 临时文件，全程不加载文件到内存，适配单日 50GB 级归档；内置 4 种算法（GCM / CBC / 两种历史格式）
• 追加式归档导出 — 按日期追加式写入归档文件：同一天只有一个归档文件，重复归档时把新增记录加密为新的自包含分段追加到文件末尾（encrypt_append），解密端逐段拼接；兼容历史单段文件
• 归档文件检查 — django_simple_archive 管理命令：list-archive-models / list-source-models 列出归档与源模型、check 逐记录验证解密与内容、archive 按源模型归档
• 后台归档任务 — 基于 django-simpletask5 的后台任务 + 核心 ArchiveButtonAdminMixin 管理后台「归档」按钮（日期选择弹窗），业务侧配置 archive_source_model 即可一键异步归档； 列表页模板链式叠加，可与 django-mptt（树形列表）、django-import-export 等 admin 扩展共存（ImportExportMixin → 业务模板 → 核心归档模板 → mptt → admin）
• 定时归档 — example 内置 cronjob（每天 00:05 归档昨天），复用 django-simpletask5 的 cron 调度器（register_cronjob + django_simpletask_crontab）与现有 worker， 无需额外定时框架
• 流式压缩中间件 — 归档先压缩后加密（信封头 compress 字段自描述，历史未 压缩文件解密时自动跳过解压），DJANGO_LOG_ARCHIVE_COMPRESSOR 可配（默认 gzip / None 关闭），临时文件根目录 DJANGO_LOG_ARCHIVE_TEMP_DIR 可配；全程流式
• 归档导出与重建 — 管理命令 export（解密 + 解压导出明文 JSONL 到指定目录， 如 OSS / 冷备盘）、reencrypt（多策略链解密旧策略文件、按当前主引擎 + 压缩器 重新加密，密钥 / 算法轮换后重建历史）、download（同步生成明文 zip 压缩包）
• 归档下载管理 — 核心数据模型 ArchiveDownloadTask（两级完成度进度 + 结果 文件），同一核心执行器支撑 django-simpletask5 后台异步与同步命令；归档记录列表 action「下载所选归档记录的明文压缩包」选定范围后台生成 zip，后台「归档下载 任务」页实时看进度并下载
• 归档记录管理页 — ArchiveRecordAdminMixin：全字段只读 + 下载密文/明文 + 异步检查；单文件解密走后台任务，避免同步接口超时
• 一键集成 — django_log_archive/__init__.py 声明 app_requires （django_static_fontawesome、django_simpletask5），配合 django_app_requires 自动展开 INSTALLED_APPS 依赖

安装

pip install django-log-archive

在 INSTALLED_APPS 中注册：

INSTALLED_APPS = [
    ...
    "django_log_archive",
    # 需要后台归档任务 / 下载管理（django-simpletask5 worker）时再加（依赖 Redis）：
    "django_simpletask5",
    # 归档按钮图标（fa-solid fa-box-archive）需要：
    "django_static_fontawesome",
]

也可通过 django_app_requires 自动展开（核心包在 __init__.py 声明了 app_requires，详见下文「业务集成方案」第 1 步）。

可选配置加密密钥与算法链：

DJANGO_LOG_ARCHIVE_ENCRYPTION_KEY = "自定义归档加密密钥"   # 缺省回退 Django SECRET_KEY
DJANGO_LOG_ARCHIVE_CIPHER = "django_log_archive.crypto.aes256gcm.AES256GCMCipher"  # 主引擎，默认
DJANGO_LOG_ARCHIVE_CIPHERS = [  # 历史解密器（算法 + 密钥）链，可选
    "django_log_archive.crypto.legacy_gcm_binary.LegacyGCMBinaryCipher",
    {"cipher": "django_log_archive.crypto.legacy_json_envelope.LegacyJSONEnvelopeCipher", "key": "旧密钥"},
]

配置项一览（含默认值）

Django settings

配置项	默认值	说明
DJANGO_LOG_ARCHIVE_ENCRYPTION_KEY	未设置 → 回退 SECRET_KEY	归档加密原始密钥，实际密钥 = sha256(raw_key)
DJANGO_LOG_ARCHIVE_CIPHER	django_log_archive.crypto.aes256gcm.AES256GCMCipher（算法 aes256-gcm）	主加密引擎（加密 + 解密），类路径字符串
DJANGO_LOG_ARCHIVE_CIPHERS	空（None，不挂历史解密器，只用主引擎解密）	历史解密器链，元素为类路径字符串或 {"cipher": 类路径, "key": 旧密钥}
DJANGO_LOG_ARCHIVE_COMPRESSOR	django_log_archive.compress.GzipCompressor（gzip，默认开启）	归档流式压缩器（类路径字符串）；设 None 关闭压缩。先压缩后加密，信封头带 compress 字段，历史未压缩文件解密时按该字段自动跳过解压
DJANGO_LOG_ARCHIVE_TEMP_DIR	/tmp	归档临时文件根目录（压缩缓冲 / 解密输出 / check 临时文件），自动创建；不可写时回退系统临时目录
DJANGO_LOG_ARCHIVE_RETENTION_DAYS	未设置（None，归档后立即删除源日志）	原始日志保存天数（按日归档、到期清理）；需要源模型有 archived_at 字段
MEDIA_ROOT（Django 内置）	项目自身默认	归档文件实际落盘目录（归档文件存 MEDIA_ROOT/archives/%Y/%m/）

归档模型声明（ArchiveRecordBase 子类类属性）

属性	默认值	说明
source_model	None	源日志模型（类或 "app_label.Model"）；为 None 时不参与归档发现
serializer	None	序列化器 callable 或源模型方法名字符串；缺省调用源模型 to_archive_line()
stats_model	None	统计模型（类或字符串）；None 则不生成统计
group_fields	()	统计分组字段（如 ("date",)）
filename_prefix	"archive"	归档文件名前缀

归档流水线参数（ArchivePipeline 构造 / run_archive / 管理命令）

参数	默认值	说明
chunk_size	2000	流水线批处理大小（取数 / 写归档 / 清理的事务粒度）
flush_size	2000	导出阶段按日聚合后刷新归档文件的记录数阈值
group_field	"date"	按日归档使用的源模型日期字段名
retention_days	None	原始日志保留天数；None 时回退 DJANGO_LOG_ARCHIVE_RETENTION_DAYS 设置，仍为空则归档后立即删除
mark_field	"archived_at"	保留策略的归档标记字段（设置 retention_days 时才生效）
use_cleanup	True	是否启用源数据清理阶段
use_stats	由 stats_model 决定（有统计模型则启用）	是否启用统计聚合阶段
since / until	None（归档全部）	归档日期范围（含），datetime.date 或 "YYYY-MM-DD"

管理命令 django_simple_archive 选项

通用命令 django_simple_archive，核心包自带、无需业务编写命令。子命令：

子命令	说明
list-archive-models	列出全部归档模型（ArchiveRecordBase 子类）及其记录数
list-source-models	列出已声明流水线的源日志模型
check	按记录校验归档文件（解密 + 内容一致性），全程流式 + 临时文件
archive	归档源日志（加密导出 + 聚合统计 + 清理源数据），支持 --mode sync|async
export	解密 + 解压归档文件，导出明文 JSONL 到指定目录（如 OSS 挂载目录 / 冷备盘）
reencrypt	历史归档按当前加密策略重新加密（密钥 / 算法轮换后重建历史文件）
download	生成归档明文 zip 压缩包（同步执行，但创建匹配的下载管理数据记录）

公共筛选选项（除 archive 外均适用）：

选项	默认值	说明
--model app_label.Model	全部模型	限定模型，可多次指定
--id N	None	只处理单条归档记录（主键）
--since / --until	None（全部）	日期范围（含），YYYY-MM-DD
--limit N	None（不限制）	每个归档模型最多处理多少条

archive 专属选项：

选项	默认值	说明
--chunk-size N	2000	批处理大小
--retention-days N	取 DJANGO_LOG_ARCHIVE_RETENTION_DAYS，仍为空则立即删除	原始日志保留天数
--mode sync|async	sync	同步执行或创建后台任务
--dry-run	False	仅统计待归档数量，不实际执行

export 专属选项：--output-dir PATH（必填，明文导出目录，自动创建）。 reencrypt 专属选项：--dry-run（只校验能否解密并预览，不实际写回）。

ArchiveButtonAdminMixin 属性

属性	默认值	说明
archive_source_model	None（缺省由归档模型声明推断）	后台归档按钮的源日志模型
archive_form_class	django_log_archive.forms.ArchiveLogsForm	归档弹窗表单
archive_verbose_name	"归档日志"	弹窗标题
archive_url_name	None → <app_label>_<model>_archive	归档视图 URL 名称
change_list_template	django_log_archive/admin/change_list.html（默认）	列表页模板；业务自定义模板须 {% extends %} 核心模板以保留归档按钮（链式叠加，见下）

后台「归档」按钮的列表页模板与静态资源均为核心包自带：

• 核心模板 django_log_archive/admin/change_list.html（按钮 + 弹窗 + 快捷日期），其 父模板按 MRO 自动解析（见「列表页模板链式叠加」），静态样式/脚本抽为 static/django_log_archive/admin/archive.css / archive.js，随 collectstatic 发布。

加解密内部参数（固定常量）

常量	默认值	说明
分块大小	65536 字节（64KB）	流式加解密按块处理，不加载文件到内存
信封头读取上限	4096 字节	解析第一行 JSON 头的最大长度
nonce / IV	16 字节	AES-GCM 初始化向量
GCM tag	16 字节	认证标签长度
归档文件存储路径	archives/%Y/%m/<文件名>	按归档日期（年月）分目录

后台异步归档（--mode async / admin 按钮 / 定时归档）还需 django-simpletask5 的 配置（如 DJANGO_SIMPLETASK_BROKER_URL，默认指向 Redis），详见 django-simpletask5 文档。

快速开始（内置 demo + example）

仓库内置 django_log_archive_demo（演示项目）与 django_log_archive_example（验证应用）。

pip install -r requirements.txt
python manage.py migrate
python manage.py generate_api_logs --days 3 --per-day 100   # 生成演示源日志
python manage.py django_simple_archive list-source-models    # 列出源日志模型
python manage.py django_simple_archive archive --until 2026-08-03   # 同步归档
python manage.py django_simple_archive archive --until 2026-08-03 --mode async   # 后台异步归档
python manage.py django_simple_archive list-archive-models   # 列出归档模型
python manage.py django_simple_archive check                 # 逐记录验证解密与内容
python manage.py django_simple_archive export --output-dir /backup/plain   # 导出明文 JSONL（如 OSS）
python manage.py django_simple_archive reencrypt --since 2026-01-01        # 按当前策略重新加密
python manage.py django_simple_archive download --since 2026-08-01         # 生成明文 zip 压缩包（同步）
python manage.py runserver                                  # 管理后台 http://127.0.0.1:8000/admin/

管理后台中打开「API 调用日志」列表，点击右上角「归档」按钮，弹出日期选择 （起始/截止日期），提交后由 django-simpletask5 在后台执行归档。

另在「API 分类」列表页可看到三者共存效果：归档按钮 + import/export 按钮 + mptt 树形缩进列表同屏显示（ApiCategoryAdmin 组合 ImportExportMixin、 ArchiveButtonAdminMixin、MPTTModelAdmin）。

如需本地跑通后台任务，需要 Redis（示例默认 redis://:changeme@127.0.0.1:6390/0） 并另开一个终端启动 worker：

python manage.py django_simpletask_executor --workers 1

运行验证测试：

python -m pytest            # 或 python manage.py test

测试结构

• django_log_archive/tests/ — 通用测试（可随扩展包复用）：加密回环 / 追加式多段 / 算法链 / 密钥派生 / 信封头解析 / 流水线分块与阶段顺序
• django_log_archive_example/tests/ — 示例相关测试：追加式归档流水线端到端 / 聚合统计 / 生成与归档管理命令 / django_simple_archive 各分支 / 后台任务与 admin 归档视图

manage.py test 与 python -m pytest 均可运行；pytest 默认附加覆盖率统计 （django_log_archive 覆盖率 ≥ 95%，见 pyproject.toml）。

项目结构

django-log-archive/
├── django_log_archive/              # 核心扩展包（打包发布）
│   ├── models.py                    # 抽象归档模型
│   ├── pipeline.py                  # 分块流水线框架（ChunkedPipeline / PipelineStage / iter_batches）
│   ├── stages.py                    # 通用归档阶段（ArchiveExportStage / SourceCleanupStage）与默认流水线 ArchivePipeline
│   ├── aggregates.py                # 统计聚合器
│   ├── checks.py                    # 归档模型发现 / 归档文件校验逻辑
│   ├── registry.py                  # 归档/源模型发现与流水线配置解析
│   ├── archive.py                   # 通用归档服务（run_archive）
│   ├── tasks.py                     # 无任务模型异步归档（submit_archive_task / ArchiveExecutionExecutor）
│   ├── admin.py                     # 后台归档按钮 mixin（ArchiveButtonAdminMixin）
│   ├── forms.py                     # 归档弹窗表单（ArchiveLogsForm）
│   ├── management/commands/         # django_simple_archive 综合管理命令
│   ├── templates/                   # 后台归档按钮核心模板（按钮 / 弹窗 / 快捷日期）
│   ├── static/                      # 归档按钮样式与脚本（archive.css / archive.js）
│   ├── crypto/                      # 流式加解密引擎（4 种算法，含追加式多段）
│   └── migrations/                  # 空（抽象模型无实质 DDL）
├── django_log_archive_demo/         # 演示项目（settings / urls / asgi / wsgi）
├── django_log_archive_example/      # 验证应用（具体模型 / 后台任务 / admin 归档按钮）
├── manage.py
├── pyproject.toml
└── README.md

模型

• ArchiveRecordBase — 归档记录基类（抽象）。字段：date/year/month/day、created_at/updated_at、 file（JSONL 归档文件，upload_to=django_log_archives）、record_count/failed_count/file_size。
• ArchiveStatsBase — 归档统计基类（抽象）。声明 aggregates（聚合器列表）并提供 build_aggregator(group_fields=())。
• ArchiveDownloadTask — 归档下载管理（核心具体模型，自带迁移）。字段：mode （bundle zip 明文压缩包 / single 单文件明文 / check 检查解密）、 archive_model_label、status（pending / running / success / failed / canceled）与 两级完成度：record_total / record_done / record_failed（记录级）+ step_total / step_done（步骤级，step_total = record_total × 2 + 1，含 zip 收尾步骤）+ current_record_id / current_step（当前记录及其子步骤）；结果文件 result_file / result_size、result_message（check 结果）、error、execution_id （simpletask5 任务）。progress_percent() 返回整体完成度百分比。
• WithArchiveDateFields / WithArchiveTimestampFields — 日期与时间戳混入。

业务应用继承抽象基类并覆写字段即可。示例见 django_log_archive_example/models.py：

class ApiLogArchive(ArchiveRecordBase):
    source_model = ApiLog                 # 源日志模型（供 archive 命令与流水线使用）
    serializer = ApiLog.to_archive_line   # 序列化器：callable 或源模型方法名字符串
    stats_model = "django_log_archive_example.ApiLogArchiveStats"
    group_fields = ("date",)
    filename_prefix = "api_log"

    class Meta:
        verbose_name = "API 日志归档"


class ApiLogArchiveStats(ArchiveStatsBase):
    call_count = models.PositiveIntegerField(default=0)
    success_count = models.PositiveIntegerField(default=0)
    failed_count = models.PositiveIntegerField(default=0)
    prompt_tokens = models.BigIntegerField(default=0)
    completion_tokens = models.BigIntegerField(default=0)
    total_tokens = models.BigIntegerField(default=0)

    aggregates = [
        CountAggregate(output_field="call_count"),
        SuccessFailureAggregate(),
        TokensAggregate(),
    ]

django_simple_archive 命令通过归档模型上声明的 source_model 等配置自动发现 源模型与流水线参数，无需业务编写命令或服务：list-source-models 列出已声明 source_model 的源模型，archive --model app_label.Model 可指定归档某个源模型。

流水线

pipeline.py 提供 ChunkedPipeline（分块批处理，CHUNK_SIZE 可配）、PipelineStage （每个阶段实现 run(input)，前序输出作为本阶段输入）与 iter_batches 工具，用于 大表按批次取数、写归档、统计、删记录的流式处理。

stages.py 内置通用阶段，业务可「选择使用哪些中间件」：

• ArchiveExportStage — 按日追加式导出：每天只有一个归档记录 / 一个归档文件， 重复归档同一日期时把新增记录加密为新的自包含分段追加到文件末尾（核心 encrypt_append），check 命令逐段解密校验；自动合并旧「分区保存」遗留的 同日重复记录，历史无分段头（旧格式）的文件走「解密既有 + 追加 + 重加密」兼容路径。
• SourceCleanupStage — 归档成功后删除已归档的源日志；process_batch 仅收集待删 id， teardown 逆序执行、导出阶段把归档文件落盘后才删除源日志，导出失败则跳过删除， 避免数据丢失。
• StatsAggregator — 分组聚合统计（见下节）。

ArchivePipeline 是集成了标准默认阶段（[SourceCleanupStage, ArchiveExportStage, StatsAggregator]）的基类，业务只需传模型与序列化器，可按需关闭默认阶段： use_cleanup / use_stats。

from django_log_archive.stages import ArchivePipeline

pipeline = ArchivePipeline(
    log_model=ApiLog,
    archive_model=ApiLogArchive,
    serializer=ApiLog.to_archive_line,        # 一行源日志 -> JSONL 行
    stats_model=ApiLogArchiveStats,
    group_fields=("date",),
    filename_prefix="api_log",
    chunk_size=2000,
)
context = pipeline.run(ids)   # context["stats"] 为分组聚合结果

serializer 缺省使用 log.to_archive_line()，也可传 callable 或模型方法名字符串。 如需自定义能力，可用 ChunkedPipeline 自行组装各阶段，选择自己需要的中间件。

按日归档、原始日志保存 N 天

默认行为是归档后立即删除源日志；需要「原始日志保存 N 天」时给 ArchivePipeline 传 retention_days=N：

• 归档时源记录打上 archived_at 标记（源模型需有该可空字段），而非立即删除；
• 之后每次归档运行时清理「已归档且归档时间早于 N 天前」的记录（purge_expired）；
• 导出阶段只处理未标记的行，保留期内重复归档不会把同一批数据写进归档文件。

# 源模型需提供 archived_at 字段
class ApiLog(models.Model):
    ...
    archived_at = models.DateTimeField(null=True, blank=True, db_index=True)

pipeline = ArchivePipeline(
    log_model=ApiLog,
    archive_model=ApiLogArchive,
    serializer=ApiLog.to_archive_line,
    stats_model=ApiLogArchiveStats,
    group_fields=("date",),
    retention_days=7,          # 原始日志保存 7 天
    chunk_size=2000,
)

示例命令可用 --retention-days 覆盖，也可全局配置默认值：

python manage.py archive_api_logs --retention-days 7
DJANGO_LOG_ARCHIVE_RETENTION_DAYS=7 python manage.py django_simple_archive archive

django-simpletask5 后台任务执行器直接复用 run_archive，保留天数缺省取自该设置。

聚合

aggregates.py 定义 BaseAggregate（collect(row) / result()）与内置聚合： CountAggregate、SuccessFailureAggregate、TokensAggregate，由 StatsAggregator 驱动对结果集分组聚合。

加解密（crypto/）

框架位于 django_log_archive/crypto/，流式加解密、全部分块处理，支持大文件与 CephFS/S3 存储。内置四种算法，每种一个文件、一个算法名：

算法名	类	文件	说明
aes256-gcm	AES256GCMCipher	aes256gcm.py	当前标准信封 AES-256-GCM 流式（默认引擎）
aes256-cbc	AES256CBCCipher	aes256cbc.py	标准信封 AES-256-CBC + PKCS#7 流式
legacy-gcm-binary	LegacyGCMBinaryCipher	legacy_gcm_binary.py	历史格式：nonce(16) + 密文 + tag(16) 裸 GCM，无信封头
legacy-json-envelope	LegacyJSONEnvelopeCipher	legacy_json_envelope.py	历史格式：{"nonce","tag","data"} JSON envelope（仅解密）

标准信封格式

加密输出为两段式（aes256-gcm / aes256-cbc）：

{"version":1,"alg":"aes256-gcm","iv":"<base64>","len":<密文载荷字节数>[,"compress":"gzip"]}\n
<流式密文 64KB 分块>
<末尾算法尾部：GCM 的 16 字节 tag / CBC 无>

头部自描述算法名与初始化向量，正文流式处理。可选字段 len 标记本段密文载荷长度， 用于追加式多段归档：一个物理文件可含多段（每段 = 头 + 密文 + 算法尾部）， decrypt_stream 逐段解密拼接；历史文件（无 len）视为单段到文件末尾，保持兼容。

流式压缩：启用压缩时（默认）在加密前对明文做 gzip 压缩（先压缩后加密， 密文不可压缩故必须压缩在前），信封头带 compress 字段；解密时按该字段自动解压， 历史未压缩文件（无该字段）自动跳过，向后兼容。压缩器可用 DJANGO_LOG_ARCHIVE_COMPRESSOR 配置（默认 GzipCompressor，None 关闭）。

密钥

key = sha256(raw_key)
raw_key 优先取 DJANGO_LOG_ARCHIVE_ENCRYPTION_KEY，缺省回退 Django SECRET_KEY。

每个引擎 = （算法 + 密钥）组合：构造时传 key 参数（已派生密钥）即产生不同密钥的 独立解密器，用于密钥轮换后解密历史文件。

使用

from django_log_archive.crypto import get_crypto

crypto = get_crypto()
crypto.encrypt_stream(src_file, dst_file)
crypto.decrypt_stream(src_file, dst_file)
crypto.encrypt_file(src_path, dst_path)
crypto.decrypt_file(src_path, dst_path)

全程流式处理（64KB 分块 + 临时文件），不加载文件内容到内存。各算法文件的 docstring 含完整格式说明与外部解密步骤。

管理命令

综合归档管理命令 django_simple_archive，支持子命令 list-archive-models / list-source-models / check / archive / export / reencrypt / download。 核心包自带，无需业务编写命令。

列出归档模型（供 check 使用）

python manage.py django_simple_archive list-archive-models

列出所有继承 ArchiveRecordBase 的具体归档模型及其记录数。

列出源日志模型（供 archive 使用）

python manage.py django_simple_archive list-source-models

列出所有在归档模型上声明了 source_model 的源日志模型及其对应归档模型。

校验归档文件可解密性

python manage.py django_simple_archive check [--model app_label.Model]... [--since YYYY-MM-DD] [--until YYYY-MM-DD] [--limit N]

• 缺省依次检查全部归档模型；--model 可多次指定，限定只检查指定模型；
• --since / --until 限定检查日期范围（全部或指定范围）；
• 每个记录分步输出检查过程：读取记录 → 发现文件 → 复制文件到临时文件（进度）→ 解密临时文件（进度）→ 检查内容提取统计字段（进度）→ 本记录完成情况；
• 复制/解密/检查全程流式（64KB 分块）与临时文件，不加载文件内容到内存，适配 单日 50GB 级归档；进度按 5% 粒度回显（文件过小时自动省略）；
• 统计字段含记录总数、有效 JSON 行、成功/失败数、prompt/completion/total tokens， 输出模型与总体汇总（真实解密成功/失败/解密字节/有效记录/成功率）。

归档（同步 / 异步）

python manage.py django_simple_archive archive [--model app_label.Model]... [--since YYYY-MM-DD] [--until YYYY-MM-DD] [--chunk-size N] [--mode sync|async] [--dry-run]

• --model 可多次指定，限定只归档指定源日志模型（源模型需在归档模型上声明 source_model）；缺省归档全部已声明的源模型；
• --mode sync（默认）：当前进程内直接执行归档（加密导出 + 聚合统计 + 清理源数据）；
• --mode async：创建后台归档任务（django_log_archive.tasks.submit_archive_task， 不绑定业务任务模型），由 django-simpletask5 的 django_simpletask_executor worker 在后台执行；
• --since / --until 限定归档日期范围（均为含，缺省归档全部）；
• --dry-run 仅统计待归档数量，不实际执行。

导出明文（export）

python manage.py django_simple_archive export --output-dir /backup/plain [--model app_label.Model]... [--id N] [--since YYYY-MM-DD] [--until YYYY-MM-DD] [--limit N]

• --output-dir 必填：明文导出目录（如 OSS 挂载目录 / 冷备盘），自动创建；
• 每个归档记录解密（自动识别压缩，历史未压缩文件自动跳过解压）后，以明文 JSONL 写出为 xxx.jsonl（密文为 xxx.jsonl.enc），不修改源数据；
• 单条解密失败时清理残留文件、记录错误并继续处理其余记录；
• 全程流式 + 临时文件，不加载文件内容到内存。

重新加密（reencrypt）

python manage.py django_simple_archive reencrypt [--model app_label.Model]... [--id N] [--since YYYY-MM-DD] [--until YYYY-MM-DD] [--limit N] [--dry-run]

• 用于密钥 / 算法轮换后把历史归档重建为新策略：解密走多策略链 （DJANGO_LOG_ARCHIVE_CIPHERS 挂旧算法 + 旧密钥，自动兼容旧策略文件）， 重新加密强制使用当前主引擎（DJANGO_LOG_ARCHIVE_CIPHER）与当前压缩器；
• 先落新文件再删旧文件，避免数据丢失；
• --dry-run 只校验能否按当前链解密并预览，不实际写回归档文件。

生成明文压缩包（download）

python manage.py django_simple_archive download [--model app_label.Model]... [--id N] [--since YYYY-MM-DD] [--until YYYY-MM-DD] [--limit N]

• 对指定范围的归档记录逐个解密（边解密边写入），生成明文 zip 压缩包；
• 同步执行，不经过 django-simpletask5 后台队列；但每个模型创建一条匹配的 下载管理数据记录（ArchiveDownloadTask，含两级完成度进度），结果 zip 作为 该记录的结果文件保存，可在后台「归档下载任务」页查看进度并下载——与后台异步 bundle 下载完全一致的数据模型。

归档下载管理（明文压缩包 / 单文件 / 异步检查）

下载管理围绕核心数据模型 ArchiveDownloadTask（含两级完成度进度字段），同一套 核心执行器同时支撑后台异步（django-simpletask5 worker）与同步命令：

• 两级进度：record_total / record_done（记录级）+ step_total / step_done （步骤级）。处理一条归档记录固定 2 个子步骤（解密解压 → 写入产物），生成 zip / 文件时再加 1 个收尾步骤，因此 step_total = record_total × 2 + 1， 压缩包这个额外步骤同样计入总数与已完成数；另有 current_record_id / current_step 指示当前处理到哪条记录、已完成几个子步骤；
• 三种模式：bundle（zip 明文压缩包）/ single（单文件明文）/ check（检查 解密，结果写入 result_message）；
• 两条路径共用：django_log_archive.download.run_download_task 是核心执行器， 后台由 ArchiveDownloadExecutor（worker）调用，命令与业务代码可直接同步调用；
• 全程流式（64KB 分块 + 临时文件），zip 内逐条记录边解密边写入压缩条目。

后台异步入口：

from django_log_archive.tasks import submit_download_task

task = submit_download_task(
    "myapp.ApiCallLogArchive",  # 归档模型（类或 "app_label.Model"）
    mode="bundle",              # bundle / single / check
    ids=[1001, 1002],           # 归档记录主键列表
)
# task.record_total / task.step_total 创建时即预置，可在 admin 查看进度

后台 worker 执行后更新该记录的 status（pending → running → success / failed）、 两级进度与结果文件（result_file，后台「归档下载任务」页提供下载链接，页面在 pending / running 时每 3 秒自动刷新）。

归档记录管理页（ArchiveRecordAdminMixin）

归档记录 ModelAdmin 继承核心 mixin 即可获得「全字段只读 + 下载密文/明文 + 检查」：

# myapp/admin.py
from django.contrib import admin
from django_log_archive.admin import ArchiveRecordAdminMixin

from .models import ApiCallLogArchive


@admin.register(ApiCallLogArchive)
class ApiCallLogArchiveAdmin(ArchiveRecordAdminMixin, admin.ModelAdmin):
    list_display = ("id", "date", "record_count", "failed_count", "file_size")

• has_change_permission 恒为 False：变更页只读（可查看 / 删除，不可编辑）；
• 归档文件字段渲染「下载密文 / 下载明文」按钮：密文直出原文件（同步）；明文走 后台异步（单文件解密可能耗时，同步接口无法在可接受时间内完成），提交 single 下载任务并跳转到「归档下载任务」页；
• 右上角 object-tools「检查」按钮提交异步 check 任务，结果写入下载管理记录的 result_message；
• 列表页提供 action「下载所选归档记录的明文压缩包」：勾选记录范围 → 提交 bundle 任务 → 后台生成明文 zip → 在「归档下载任务」页下载。

后台归档任务（django-simpletask5）

核心 django_log_archive.tasks 提供无任务模型的异步归档：submit_archive_task 直接创建 TaskExecution（参数存于 context，executor_class 指向核心通用执行器 ArchiveExecutionExecutor）并发布消息，worker 调用执行器完成归档并更新任务状态—— 业务无需定义任何 django-simpletask5 Task 子类。

管理后台「API 调用日志」列表页的「归档」按钮（位于「添加」按钮之前）由核心 ArchiveButtonAdminMixin 提供，弹出起始/截止日期选择弹窗，提交后创建后台任务 并在列表顶部提示；任务执行状态可在 django-simpletask5 自带的「任务执行」管理页 查看（执行器 / 状态 / 重试等）。

业务侧复用归档按钮只需在 ModelAdmin 上继承 mixin 并配置源日志模型：

from django_log_archive.admin import ArchiveButtonAdminMixin


@admin.register(ApiLog)
class ApiLogAdmin(ArchiveButtonAdminMixin, admin.ModelAdmin):
    archive_source_model = ApiLog                 # 源日志模型（也可由归档模型声明推断）
    archive_verbose_name = "归档 API 调用日志"     # 可选：弹窗标题
    # 可选：archive_form_class（缺省核心 ArchiveLogsForm）、archive_url_name

列表页模板链式叠加：mixin 默认声明 change_list_template = django_log_archive/admin/change_list.html（核心归档模板，自带按钮 + 弹窗 + JS）。 业务自定义模板优先、核心模板缺省兜底：业务若因其它需求重载 change_list.html，只需 让自己的模板以核心模板为父模板，并在需要叠加内容的 block 里用 {{ block.super }} 保留归档按钮；多个业务模板可逐层 {% extends %} 相互扩展（内容逐层叠加），互不覆盖：

{# myapp/templates/myapp/admin/change_list.html #}
{% extends "django_log_archive/admin/change_list.html" %}
{% block object-tools-items %}
  <li><a href="#" class="my-custom-btn">自定义按钮</a></li>
  {{ block.super }}   {# 保留归档按钮 #}
{% endblock %}

class ApiLogAdmin(ArchiveButtonAdminMixin, admin.ModelAdmin):
    change_list_template = "myapp/admin/change_list.html"   # 自定义模板照常设置

核心模板的父模板自动解析：{% extends %} 只能有一个父模板，而 mptt / grappelli 等第三方扩展也会声明自己的 change_list_template（MRO 里只有一个能胜出）。mixin 通过 archive_base_change_list_template 上下文变量决定核心模板的父模板—— _get_archive_base_change_list_template() 沿 MRO 查找第一个非本 mixin（也非当前 生效模板）的 change_list_template：继承 MPTTModelAdmin 时自动命中 admin/mptt_change_list.html（树形列表），grappelli 等其它自定义 base admin 同理 自动命中，无需逐个适配；找不到则回退 admin/change_list.html。

与 import-export / mptt 三者共存（example 内置演示）：django_log_archive_example 新增 ApiCategory（MPTT 树形模型）与 ApiCategoryAdmin，在同一列表页同时展示 归档按钮 + import/export 按钮 + mptt 树形缩进列表。mixin 顺序固定为 ImportExportMixin 在最前（它会把 change_list_template 作为 base_change_list_template 包装成 import-export 模板）、ArchiveButtonAdminMixin 居中、MPTTModelAdmin 殿后，渲染链为 「import/export → 业务模板 → 核心归档模板 → mptt → admin」：

from import_export import resources
from import_export.admin import ImportExportMixin
from mptt.admin import MPTTModelAdmin

from django_log_archive.admin import ArchiveButtonAdminMixin
from .models import ApiCategory, ApiLog


class ApiCategoryResource(resources.ModelResource):
    class Meta:
        model = ApiCategory
        fields = ("id", "name", "parent")


@admin.register(ApiCategory)
class ApiCategoryAdmin(
    ImportExportMixin,
    ArchiveButtonAdminMixin,
    MPTTModelAdmin,
):
    resource_classes = [ApiCategoryResource]
    archive_source_model = ApiLog        # 归档按钮点击即归档 ApiLog
    list_display = ("name", "parent")    # mptt 自动对首字段做树形缩进

注意：demo 项目的 INSTALLED_APPS 需加入 mptt 与 import_export（第三方模板靠 AppDirectoriesFinder 从这两者加载）。

后台依赖 Redis（消息队列 + 分布式锁），启动 worker：

python manage.py django_simpletask_executor --workers 1

定时归档（example 内置示例）

example 通过 django-simpletask5 的 cron 调度器注册了「每天 00:05 归档昨天」的 cronjob（django_log_archive_example/cronjobs.py）：

• DailyApiLogArchiveExecutor — 执行时动态计算「昨天」（timezone.localdate() - 1day）， 调用核心通用归档服务 run_archive 并更新执行状态；「昨天」不写入 cronjob 静态 context，也可用 context 的 data 覆盖（since / until / chunk_size / retention_days）；
• register_example_cronjobs() — 在 AppConfig.ready() 中调用，向 django-simpletask5 内存注册表写入 cronjob（cron 表达式 5 0 * * *）；调度器每轮 tick 自动 sync_cronjobs() 同步到数据库（django_simpletask5_cronjob 表，也可 用 django_simpletask_sync_cronjobs 命令手动同步）。

运行定时归档需要 Redis，并常驻两个进程（调度器 + worker）：

python manage.py django_simpletask_crontab                        # 定时调度器
python manage.py django_simpletask_executor --workers 1           # 任务 worker

cronjob 可在后台「任务计划（CronJob）」管理页查看 / 启停；业务侧复用只需在 自己的 app 中 register_cronjob(...) 指向一个执行器，并在 AppConfig.ready() 中调用即可。若不想常驻调度器进程，也可改用 OS crontab 直接调用 django_simple_archive archive --until $(date -d yesterday +%F)。

业务集成方案（以 example 为例）

django_log_archive_example 是一份可直接对照的完整业务实现：源模型（ApiLog）、 归档模型（ApiLogArchive）、统计模型（ApiLogArchiveStats）、admin 归档按钮、业务 归档服务（archive.py）、定时归档（cronjobs.py）齐全。下面按步骤说明如何把 归档能力接入你自己的业务 app（下文以 myapp 为例）。

1. 安装与配置

pip install django-log-archive

INSTALLED_APPS = [
    ...
    "django_log_archive",
    # 需要后台归档任务 / 下载管理（django-simpletask5 worker）时再加（依赖 Redis）：
    "django_simpletask5",
    # 归档按钮图标（fa-solid fa-box-archive）需要：
    "django_static_fontawesome",
]

也可用 django_app_requires 自动展开依赖（核心包在 django_log_archive/__init__.py 声明了 app_requires，含 django_static_fontawesome 与 django_simpletask5）：

# settings.py
INSTALLED_APPS = [...]
from django_app_requires import patch_all
patch_all()          # 自动把 app_requires 依赖展开进 INSTALLED_APPS

DJANGO_LOG_ARCHIVE_ENCRYPTION_KEY = "自定义归档加密密钥"   # 缺省回退 Django SECRET_KEY
DJANGO_LOG_ARCHIVE_CIPHER = "django_log_archive.crypto.aes256gcm.AES256GCMCipher"  # 默认
# DJANGO_LOG_ARCHIVE_COMPRESSOR = "django_log_archive.compress.GzipCompressor"  # 默认开启压缩
# DJANGO_LOG_ARCHIVE_COMPRESSOR = None   # 关闭压缩（历史未压缩文件仍可正常解密）
# DJANGO_LOG_ARCHIVE_TEMP_DIR = "/mnt/big-disk/tmp"   # 归档临时文件根目录（默认 /tmp）

2. 定义源日志模型

源模型需有按日归档的日期字段（example 用 date），如需「原始日志保留 N 天」 策略则加可空 archived_at 字段，并提供一行转 JSON 的序列化方法：

# myapp/models.py
class ApiCallLog(models.Model):
    date = models.DateField(db_index=True)
    is_success = models.BooleanField(default=True)
    total_tokens = models.PositiveIntegerField(default=0)
    payload = models.TextField(blank=True, default="")
    archived_at = models.DateTimeField(null=True, blank=True, db_index=True)  # 可选：保留策略

    def to_archive_line(self):
        import json
        return json.dumps({
            "id": self.pk,
            "date": self.date.isoformat(),
            "is_success": self.is_success,
            "total_tokens": self.total_tokens,
            "payload": self.payload,
        }, ensure_ascii=False)

3. 定义归档模型与统计模型（核心一步）

继承 ArchiveRecordBase 的归档模型是整个集成的心脏：所有流水线配置都声明在 上面（django_simple_archive 命令、admin 按钮、run_archive 均据此自动发现与 执行），业务无需编写命令或服务：

# myapp/models.py
from django_log_archive.aggregates import CountAggregate, SuccessFailureAggregate
from django_log_archive.models import ArchiveRecordBase, ArchiveStatsBase


class ApiCallLogArchive(ArchiveRecordBase):
    source_model = ApiCallLog               # 源日志模型（类或 "app_label.Model"）
    serializer = ApiCallLog.to_archive_line # 序列化器：callable 或方法名字符串
    stats_model = "myapp.ApiCallLogArchiveStats"  # 统计模型，可省略
    group_fields = ("date",)                # 统计分组字段
    filename_prefix = "api_call"            # 归档文件名前缀

    class Meta:
        verbose_name = "API 调用日志归档"


class ApiCallLogArchiveStats(ArchiveStatsBase):
    call_count = models.PositiveIntegerField(default=0)
    success_count = models.PositiveIntegerField(default=0)
    failed_count = models.PositiveIntegerField(default=0)

    aggregates = [
        CountAggregate(output_field="call_count"),
        SuccessFailureAggregate(),
    ]

完成后执行 python manage.py makemigrations myapp && python manage.py migrate， 并可用核心命令验证配置已生效：

python manage.py django_simple_archive list-source-models    # 应看到 myapp.ApiCallLog
python manage.py django_simple_archive list-archive-models   # 应看到 myapp.ApiCallLogArchive

4. 归档触发方式（任选其一或组合）

A. 管理命令（无需任何代码） — 核心自带，同步/异步/校验：

python manage.py django_simple_archive archive --until 2026-08-03            # 同步
python manage.py django_simple_archive archive --until 2026-08-03 --dry-run  # 预统计
python manage.py django_simple_archive archive --until 2026-08-03 --mode async  # 后台任务
python manage.py django_simple_archive check                                 # 校验归档文件

B. 业务代码调用 run_archive — 在视图/脚本/管理命令里直接归档（example 的 django_log_archive_example/archive.py 就是给它包了一层「固定归档 ApiLog」的便捷 签名，非必需）：

from django_log_archive.archive import run_archive

result = run_archive(
    "myapp.ApiCallLog",        # 源模型类或 "app_label.Model" 字符串
    since="2026-08-01",
    until="2026-08-31",
    chunk_size=2000,
    retention_days=7,
)
# result == {"source_count": ..., "archive_count": ..., "stats_count": ..., "remaining": ...}

C. 管理后台「归档」按钮 — ModelAdmin 继承核心 mixin 即可，弹出日期选择弹窗， 提交后由 worker 异步执行（example 的 admin.py）：

# myapp/admin.py
from django.contrib import admin
from django_log_archive.admin import ArchiveButtonAdminMixin

from .models import ApiCallLog


@admin.register(ApiCallLog)
class ApiCallLogAdmin(ArchiveButtonAdminMixin, admin.ModelAdmin):
    archive_source_model = ApiCallLog            # 也可省略，由归档模型声明推断
    archive_verbose_name = "归档 API 调用日志"

D. 定时归档 — 复用 django-simpletask5 cron 调度器，每天 00:05 自动归档昨天 （参照 example 的 cronjobs.py，执行时动态计算「昨天」）：

# myapp/cronjobs.py
import json
from datetime import timedelta
from django.utils import timezone
from django_log_archive.archive import run_archive


class DailyApiCallLogArchiveExecutor:
    timeout_seconds = 3600

    def execute(self, execution, task=None):
        yesterday = timezone.localdate() - timedelta(days=1)
        result = run_archive("myapp.ApiCallLog", since=yesterday, until=yesterday)
        execution.status = "success"
        execution.finished_at = timezone.now()
        execution.save(update_fields=["status", "finished_at", "updated_at"])
        return json.dumps({"success": True, "date": yesterday.isoformat(), **result}, ensure_ascii=False)


def register_myapp_cronjobs():
    from django_simpletask5.core.cronjob_registry import register_cronjob
    register_cronjob(
        uid="myapp_daily_api_call_log_archive",
        display_name="每日归档昨日 API 调用日志",
        cron_expression="5 0 * * *",
        executor_class="myapp.cronjobs.DailyApiCallLogArchiveExecutor",
    )

# myapp/apps.py
class MyappConfig(AppConfig):
    ...
    def ready(self):
        from .cronjobs import register_myapp_cronjobs
        register_myapp_cronjobs()

常驻运行：python manage.py django_simpletask_crontab（调度器）+ python manage.py django_simpletask_executor --workers 1（worker）。

5. 归档记录管理页（只读 + 下载密文/明文 + 检查）

归档记录的 ModelAdmin 继承 ArchiveRecordAdminMixin，变更页全字段只读， 归档文件字段变为「下载密文 / 下载明文」按钮，右上角多一个「检查」按钮，列表页 多一个「下载所选归档记录的明文压缩包」action（详见上文「归档记录管理页」）：

# myapp/admin.py
from django.contrib import admin
from django_log_archive.admin import ArchiveRecordAdminMixin

from .models import ApiCallLogArchive


@admin.register(ApiCallLogArchive)
class ApiCallLogArchiveAdmin(ArchiveRecordAdminMixin, admin.ModelAdmin):
    list_display = ("id", "date", "record_count", "failed_count", "file_size")

6. 归档下载管理（明文压缩包 / 单文件 / 异步检查）

核心自带 ArchiveDownloadTask 下载管理模型与后台「归档下载任务」管理页 （进度条 + 结果文件下载 + 自动刷新），无需业务注册。业务侧两种接入方式：

A. 后台异步（推荐） — 记录列表 action / 记录页「下载明文」/「检查」按钮都会 自动创建下载任务并由 django-simpletask5 worker 执行（见上文「归档下载管理」）； 业务代码也可直接调用：

from django_log_archive.tasks import submit_download_task

task = submit_download_task("myapp.ApiCallLogArchive", mode="bundle", ids=[1001, 1002])

B. 同步命令 — django_simple_archive download 在进程内同步执行，但同样创建 匹配的下载管理数据记录：

python manage.py django_simple_archive download --since 2026-08-01 --until 2026-08-31

7. 原始日志保留策略

默认归档后立即删除源日志。需要「原始日志保存 N 天」时传 retention_days=N（源 模型需有 archived_at 字段）：归档打标记而非删除，之后每次归档清理到期记录， 重复归档不产生重复数据。全局默认值可配：

DJANGO_LOG_ARCHIVE_RETENTION_DAYS=7 python manage.py django_simple_archive archive

8. 运维与校验

• 校验归档文件 — django_simple_archive check 逐记录复制/解密/比对内容，全程 流式 + 临时文件，--model / --since / --until / --limit 可限定范围抽查；
• 导出明文（如 OSS / 冷备盘） — django_simple_archive export --output-dir ... 解密 + 解压导出明文 JSONL（xxx.jsonl），不修改源数据，单条失败自动跳过；
• 密钥 / 算法轮换 — django_simple_archive reencrypt 按当前主引擎重新加密： 解密走 DJANGO_LOG_ARCHIVE_CIPHERS 多策略链（挂旧算法 + 旧密钥），加密强制 当前主引擎 + 压缩器，先落新文件再删旧文件；
• 归档可观测 — 归档记录（ApiLogArchive）含日期、记录数、失败数、文件大小； 统计模型（ApiLogArchiveStats）按 group_fields 聚合，可在 admin 或报表中展示；
• 下载管理 — 明文 zip / 单文件 / 检查统一走 ArchiveDownloadTask，后台 「归档下载任务」页查看两级进度并下载结果。

开发与打包

pip install -r requirements.txt        # Django + pycryptodome + simpletask5 + 测试依赖
python -m pytest                       # 运行验证测试（覆盖率 ≥ 95%）
python manage.py test                  # Django 自带测试运行器
python -m ruff check .                 # 代码检查
python -m build                        # 构建 sdist + wheel

打包仅发布 django_log_archive 核心包（pyproject.toml 中已排除 demo / example）。

更新记录

v0.1.0

1. 日志归档开箱即用：业务表只需继承现成的归档/统计基类并声明 source_model、序列化器等少量配置， 即可获得完整的「按日加密归档 + 统计」能力，无需自己设计归档表与写入逻辑
2. 海量数据也稳妥：归档按批次流式处理、内存占用恒定，单日 GB 级日志也能平稳归档；同一天的数据在 多次归档运行间自动追加合并，不会重复归档
3. 归档即加密存储：归档文件采用流式加密（AES-256-GCM / CBC，并兼容历史算法与旧密钥），密钥或算法 轮换后旧文件仍可正常读取
4. 省空间不牺牲兼容：归档先压缩后加密，文件自带压缩标记；早期的未压缩归档文件解压时自动识别跳过， 全程不需要把整份文件加载进内存
5. 原始日志可留可删，你说了算：归档后默认立即清理源日志释放空间；需要回溯时开启「保留 N 天」， 到期自动清理，且同一批数据不会因再次运行而被重复归档
6. 失败不丢数据：归档过程中任何一步出错，源日志都会原样保留，修复后直接重跑即可，杜绝 「归档失败却把源数据删了」的意外
7. 统计始终准确：同一天在后续归档运行中补充的数据，统计会自动累加，与归档文件里的真实内容一致
8. 后台一键归档：为源日志的 ModelAdmin 挂一个 mixin，列表页即有「归档」按钮 + 日期范围弹窗 + 异步后台任务，并能与 mptt / django-import-export / grappelli 等既有 admin 扩展共存
9. 明文随时可取：后台可把指定归档记录批量下载为明文 zip、单文件明文，或发起解密检查并逐条查看 解密结果；下载任务显示实时进度、完成后可直接下载结果文件，反复重跑也不会留下冗余文件
10. 一条命令管理全部归档：内置 django_simple_archive 管理命令，支持按模型列出归档/源模型、 逐条校验归档文件能否解密、归档、导出明文 JSONL、按新密钥策略重新加密旧文件、同步生成明文 zip
11. 随包附赠可运行演示：内置 demo 与 example 应用（含每日定时归档、归档按钮与三方 admin 共存的 完整示例），并附带 260+ 项验证测试（覆盖率 ≥ 95%），可直接照抄接入自己的项目