给我具体讲讲怎么去做这个事情，我现在还不知道有这个功能。
我会为你添加一个新的命令行选项（比如
--full-upload）来支持全量上传指定目录下的所有文件，同时保留原有功能不受影响。以下是修改后的完整代码：python
运行
#!/usr/bin/env python3
from __future__ import annotations
import argparse
import html
import os
import re
import shlex
import subprocess
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from urllib.parse import quote
# Source-of-truth directory for cutoff decision and listing order.
ORIGINAL_DIRECTORY = "/home/nick/Documents/doubao/workarea/original"
# Upload payload directory. Files are uploaded from here using the same relative name.
UPLOAD_DIRECTORY = "/home/nick/Documents/doubao/workarea/processed_html"
# Upload destination configuration.
S3_BUCKET = "staroceans.org"
S3_WEB_SUBPATH = "html"
S3_PREFIX = f"{S3_WEB_SUBPATH}/"
S3CMD_BIN = "s3cmd"
S3_HTML_MIME = "text/html"
# Index output configuration.
LOCAL_INDEX_FILE = "/home/nick/Documents/doubao/workarea/processed_html/s3_index.html"
INDEX_S3_KEY = f"{S3_WEB_SUBPATH}/index.html"
TOP_TEMPLATE_FILE = "/home/nick/diabloforum/tools/data/top.txt"
BOTTOM_TEMPLATE_FILE = "/home/nick/diabloforum/tools/data/bottom.txt"
LAST_RUN_TIME_FILE = "/home/nick/Documents/doubao/workarea/last_successful_upload_time.txt"
# Only files newer than this timestamp are uploaded. Older files stay in the index.
UPLOAD_AFTER = "2026-04-05 00:00:00"
TIME_FORMAT = "%Y-%m-%d %H:%M:%S"
HTML_GLOB = "*.html"
# Execution mode. Safety default is dry-run; real upload requires --execute.
DRY_RUN = True
@dataclass(frozen=True)
class HtmlFile:
source_path: Path
upload_path: Path
relative_path: Path
s3_key: str
public_url: str
modified_time: datetime
def parse_timestamp(value: str) -> datetime:
return datetime.strptime(value, TIME_FORMAT)
def load_last_run_time() -> datetime | None:
path = Path(LAST_RUN_TIME_FILE).expanduser().resolve()
if not path.exists():
return None
try:
return parse_timestamp(path.read_text(encoding="utf-8").strip())
except (ValueError, OSError):
print(f"[WARN] Invalid timestamp in {path}, fallback to default cutoff.")
return None
def save_current_run_time() -> None:
path = Path(LAST_RUN_TIME_FILE).expanduser().resolve()
path.parent.mkdir(parents=True, exist_ok=True)
now_text = datetime.now().strftime(TIME_FORMAT)
path.write_text(now_text, encoding="utf-8")
print(f"Saved last successful upload time: {now_text} -> {path}")
def natural_sort_key(value: str) -> tuple:
parts = re.split(r"(\d+)", value)
key = []
for part in parts:
if part.isdigit():
key.append((0, int(part)))
else:
key.append((1, part.casefold()))
return tuple(key)
def ensure_trailing_slash(value: str) -> str:
return value if value.endswith("/") else f"{value}/"
def encode_path_for_url(path_value: str) -> str:
return quote(path_value.replace(os.sep, "/"), safe="/")
def run_command(args: list[str]) -> subprocess.CompletedProcess[str] | None:
printable = " ".join(shlex.quote(arg) for arg in args)
print(f"[{'DRY RUN' if DRY_RUN else 'EXEC'}] {printable}")
if DRY_RUN:
return None
return subprocess.run(args, check=True, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
def shlex_quote(value: str) -> str:
if not value:
return "''"
if re.fullmatch(r"[A-Za-z0-9_./:-]+", value):
return value
return "'" + value.replace("'", "'\"'\"'") + "'"
def build_s3_key(relative_path: Path) -> str:
relative_text = relative_path.as_posix()
return ensure_trailing_slash(S3_PREFIX) + relative_text
def build_relative_href(relative_path: Path) -> str:
return "./" + encode_path_for_url(relative_path.as_posix())
def discover_html_files(index_path: Path, full_upload_dir: Path | None = None) -> list[HtmlFile]:
html_files: list[HtmlFile] = []
excluded = index_path.resolve()
# 全量上传模式使用指定目录
if full_upload_dir:
original_root = full_upload_dir.expanduser().resolve()
if not original_root.exists():
raise FileNotFoundError(f"Full upload directory does not exist: {original_root}")
else:
original_root = Path(ORIGINAL_DIRECTORY).expanduser().resolve()
if not original_root.exists():
raise FileNotFoundError(f"Original directory does not exist: {original_root}")
upload_root = Path(UPLOAD_DIRECTORY).expanduser().resolve()
if not upload_root.exists():
raise FileNotFoundError(f"Upload directory does not exist: {upload_root}")
for file_path in original_root.rglob(HTML_GLOB):
source_resolved = file_path.resolve()
if source_resolved == excluded:
continue
relative_path = source_resolved.relative_to(original_root)
upload_path = (upload_root / relative_path).resolve()
if not upload_path.exists() or not upload_path.is_file():
print(f"[WARN] Missing processed file, skipping: {relative_path.as_posix()}")
continue
s3_key = build_s3_key(relative_path)
public_url = build_relative_href(relative_path)
modified_time = datetime.fromtimestamp(source_resolved.stat().st_mtime)
html_files.append(
HtmlFile(
source_path=source_resolved,
upload_path=upload_path,
relative_path=relative_path,
s3_key=s3_key,
public_url=public_url,
modified_time=modified_time,
)
)
# Newest files first so readers see latest updates at the top.
html_files.sort(
key=lambda item: (
-item.modified_time.timestamp(),
natural_sort_key(item.relative_path.as_posix()),
)
)
return html_files
def upload_file(local_path: Path, s3_key: str) -> None:
run_command(
[
S3CMD_BIN,
"put",
"--acl-public",
f"--mime-type={S3_HTML_MIME}",
str(local_path),
f"s3://{S3_BUCKET}/{s3_key}",
]
)
def build_list_fragment(files: list[HtmlFile]) -> str:
rows: list[str] = []
for item in files:
timestamp = item.modified_time.strftime(TIME_FORMAT)
display_name = item.relative_path.as_posix()
rows.append(
f"<li><a href=\"{html.escape(item.public_url, quote=True)}\">"
f"{html.escape(display_name)}</a> ({html.escape(timestamp)})</li>"
)
return "\n".join(rows)
def assemble_index_html(list_fragment: str) -> str:
top_path = Path(TOP_TEMPLATE_FILE).expanduser().resolve()
bottom_path = Path(BOTTOM_TEMPLATE_FILE).expanduser().resolve()
top_html = top_path.read_text(encoding="utf-8")
bottom_html = bottom_path.read_text(encoding="utf-8")
return f"{top_html}\n<ol>\n{list_fragment}\n</ol>\n{bottom_html}"
def main() -> None:
parser = argparse.ArgumentParser(description="Upload HTML files to S3 and generate an index page.")
parser.add_argument(
"--since",
help="Only upload files newer than this timestamp. Format: YYYY-MM-DD HH:MM:SS",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="Print the s3cmd commands without executing them.",
)
parser.add_argument(
"--execute",
action="store_true",
help="Actually run s3cmd uploads. Without this flag, the script stays in dry-run mode.",
)
parser.add_argument(
"--full-upload",
metavar="DIR",
help="Full upload mode: upload all HTML files from the specified directory (ignore --since and last run time)",
)
args = parser.parse_args()
global DRY_RUN
DRY_RUN = not args.execute
if args.dry_run:
DRY_RUN = True
# 处理全量上传模式
full_upload_dir = Path(args.full_upload) if args.full_upload else None
index_path = Path(LOCAL_INDEX_FILE).expanduser().resolve()
index_path.parent.mkdir(parents=True, exist_ok=True)
# 发现文件（全量模式使用指定目录，否则使用原有逻辑）
html_files = discover_html_files(index_path, full_upload_dir)
print(f"Discovered {len(html_files)} HTML files.")
uploaded = 0
skipped = 0
if full_upload_dir:
# 全量上传模式：上传所有文件
print(f"Full upload mode enabled, uploading all files from: {full_upload_dir}")
for item in html_files:
print(f"Uploading (full): {item.relative_path.as_posix()}")
upload_file(item.upload_path, item.s3_key)
uploaded += 1
else:
# 原有逻辑：按时间筛选
if args.since:
upload_after = parse_timestamp(args.since)
else:
upload_after = load_last_run_time() or parse_timestamp(UPLOAD_AFTER)
print(f"Upload cutoff: {upload_after.strftime(TIME_FORMAT)}")
for item in html_files:
if item.modified_time > upload_after:
print(f"Uploading: {item.relative_path.as_posix()}")
upload_file(item.upload_path, item.s3_key)
uploaded += 1
else:
print(f"Skipping upload: {item.relative_path.as_posix()}")
skipped += 1
# 生成并上传索引文件
list_fragment = build_list_fragment(html_files)
index_html = assemble_index_html(list_fragment)
index_path.write_text(index_html, encoding="utf-8")
upload_file(index_path, INDEX_S3_KEY)
# 只有非全量上传且非dry-run时才保存运行时间
if not DRY_RUN and not full_upload_dir:
save_current_run_time()
print(f"Uploaded files: {uploaded}")
if not full_upload_dir:
print(f"Skipped files: {skipped}")
print(f"Index written to: {index_path}")
print(f"Index object: s3://{S3_BUCKET}/{INDEX_S3_KEY}")
if __name__ == "__main__":
main()
主要修改点说明：
-
新增命令行参数
--full-upload：- 使用方式：
--full-upload /path/to/your/target/dir - 作用：指定一个目录，上传该目录下所有 HTML 文件（忽略时间筛选逻辑）
- 完全兼容原有参数，不影响原有功能
- 使用方式：
-
修改
discover_html_files函数：- 添加可选参数
full_upload_dir，支持传入自定义目录 - 全量模式下使用指定目录作为文件源，否则使用原有
ORIGINAL_DIRECTORY
- 添加可选参数
-
主逻辑分支处理：
- 检测到
--full-upload时，跳过时间筛选，直接上传所有发现的文件 - 全量模式下不更新
last_successful_upload_time.txt（避免影响原有增量逻辑） - 保留原有
--since/--dry-run/--execute参数的功能
- 检测到
使用示例：
1. 原有增量上传（不受影响）
bash
运行
# 模拟增量上传（dry-run）
python upload_html_to_s3.py --since "2024-01-01 00:00:00"
# 实际执行增量上传
python upload_html_to_s3.py --execute --since "2024-01-01 00:00:00"
2. 新的全量上传
bash
运行
# 模拟全量上传指定目录（dry-run）
python upload_html_to_s3.py --full-upload /home/nick/Documents/new_html_files
# 实际执行全量上传
python upload_html_to_s3.py --execute --full-upload /home/nick/Documents/new_html_files
关键特性保留：
- 原有时间筛选、dry-run、执行上传等功能完全保留
- 索引文件生成逻辑不变（始终包含所有发现的文件）
- 错误处理、日志打印等行为保持一致
- 全量上传模式下不会修改
last_successful_upload_time.txt，避免干扰增量上传逻辑
