feat: 实现批量 Excel 生成(文本、图片、表格区域)
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
504bd2f65f
commit
eb82650c03
@ -1 +1,63 @@
|
||||
# placeholder
|
||||
import re
|
||||
import os
|
||||
import shutil
|
||||
from openpyxl import load_workbook
|
||||
from openpyxl.drawing.image import Image as XLImage
|
||||
|
||||
PLACEHOLDER_RE = re.compile(r"\{\{(.+?)\}\}")
|
||||
|
||||
def _apply_filename(pattern: str, row: dict) -> str:
|
||||
def replace(m):
|
||||
return str(row.get(m.group(1), m.group(0)))
|
||||
return PLACEHOLDER_RE.sub(replace, pattern)
|
||||
|
||||
def generate(req: dict) -> dict:
|
||||
template_path = req["template_path"]
|
||||
fields = req["fields"]
|
||||
rows = req.get("rows")
|
||||
data_file_path = req.get("data_file_path")
|
||||
output_dir = req["output_dir"]
|
||||
filename_pattern = req["filename_pattern"]
|
||||
|
||||
if data_file_path and not rows:
|
||||
dwb = load_workbook(data_file_path, data_only=True)
|
||||
dws = dwb.active
|
||||
headers = [cell.value for cell in next(dws.iter_rows(min_row=1, max_row=1))]
|
||||
rows = []
|
||||
for row in dws.iter_rows(min_row=2, values_only=True):
|
||||
rows.append({headers[i]: v for i, v in enumerate(row) if i < len(headers)})
|
||||
|
||||
os.makedirs(output_dir, exist_ok=True)
|
||||
results = []
|
||||
|
||||
for i, row in enumerate(rows):
|
||||
filename = _apply_filename(filename_pattern, row)
|
||||
output_path = os.path.join(output_dir, filename)
|
||||
shutil.copy2(template_path, output_path)
|
||||
|
||||
try:
|
||||
wb = load_workbook(output_path)
|
||||
for field in fields:
|
||||
value = row.get(field["name"])
|
||||
if value is None:
|
||||
continue
|
||||
ws = wb[field["sheet"]]
|
||||
if field["type"] == "text":
|
||||
ws[field["cell"]] = value
|
||||
elif field["type"] == "image":
|
||||
if os.path.exists(str(value)):
|
||||
img = XLImage(str(value))
|
||||
ws.add_image(img, field["cell"])
|
||||
elif field["type"] == "table_range":
|
||||
start_cell = ws[field["cell"]]
|
||||
start_row = start_cell.row
|
||||
start_col = start_cell.column
|
||||
for r_idx, data_row in enumerate(value):
|
||||
for c_idx, cell_value in enumerate(data_row):
|
||||
ws.cell(row=start_row + r_idx, column=start_col + c_idx, value=cell_value)
|
||||
wb.save(output_path)
|
||||
results.append({"row": i, "status": "success", "file": output_path})
|
||||
except Exception as e:
|
||||
results.append({"row": i, "status": "error", "error": str(e), "file": output_path})
|
||||
|
||||
return {"results": results}
|
||||
|
||||
4
tests/create_image_fixture.py
Normal file
4
tests/create_image_fixture.py
Normal file
@ -0,0 +1,4 @@
|
||||
from PIL import Image as PILImage
|
||||
img = PILImage.new("RGB", (100, 100), color=(255, 0, 0))
|
||||
img.save("tests/fixtures/sample_image.png")
|
||||
print("image fixture created")
|
||||
BIN
tests/fixtures/sample_image.png
vendored
Normal file
BIN
tests/fixtures/sample_image.png
vendored
Normal file
Binary file not shown.
|
After Width: | Height: | Size: 287 B |
@ -1 +1,73 @@
|
||||
# placeholder
|
||||
import sys, os, tempfile
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "../../python"))
|
||||
|
||||
from generator import generate
|
||||
from openpyxl import load_workbook
|
||||
|
||||
FIXTURE = os.path.join(os.path.dirname(__file__), "../fixtures/sample_template.xlsx")
|
||||
IMAGE = os.path.join(os.path.dirname(__file__), "../fixtures/sample_image.png")
|
||||
|
||||
def _base_request(output_dir):
|
||||
return {
|
||||
"template_path": FIXTURE,
|
||||
"fields": [
|
||||
{"name": "编号", "type": "text", "sheet": "Sheet1", "cell": "B3"},
|
||||
{"name": "姓名", "type": "text", "sheet": "Sheet1", "cell": "C5"},
|
||||
],
|
||||
"rows": [
|
||||
{"编号": "001", "姓名": "张三"},
|
||||
{"编号": "002", "姓名": "李四"},
|
||||
],
|
||||
"output_dir": output_dir,
|
||||
"filename_pattern": "{{编号}}_报告.xlsx",
|
||||
}
|
||||
|
||||
def test_generates_correct_number_of_files():
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
result = generate(_base_request(tmpdir))
|
||||
assert len(result["results"]) == 2
|
||||
|
||||
def test_output_files_exist():
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
result = generate(_base_request(tmpdir))
|
||||
for r in result["results"]:
|
||||
assert r["status"] == "success"
|
||||
assert os.path.exists(r["file"])
|
||||
|
||||
def test_text_fields_are_filled():
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
result = generate(_base_request(tmpdir))
|
||||
wb = load_workbook(result["results"][0]["file"])
|
||||
ws = wb["Sheet1"]
|
||||
assert ws["B3"].value == "001"
|
||||
assert ws["C5"].value == "张三"
|
||||
|
||||
def test_filename_pattern_applied():
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
result = generate(_base_request(tmpdir))
|
||||
names = [os.path.basename(r["file"]) for r in result["results"]]
|
||||
assert "001_报告.xlsx" in names
|
||||
assert "002_报告.xlsx" in names
|
||||
|
||||
def test_missing_field_is_skipped_not_crashed():
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
req = _base_request(tmpdir)
|
||||
req["rows"] = [{"编号": "003"}] # 缺 姓名
|
||||
result = generate(req)
|
||||
assert result["results"][0]["status"] == "success"
|
||||
|
||||
def test_image_field_is_inserted():
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
req = {
|
||||
"template_path": FIXTURE,
|
||||
"fields": [
|
||||
{"name": "图片1", "type": "image", "sheet": "Sheet1", "cell": "A1"},
|
||||
],
|
||||
"rows": [{"图片1": IMAGE}],
|
||||
"output_dir": tmpdir,
|
||||
"filename_pattern": "img_test.xlsx",
|
||||
}
|
||||
result = generate(req)
|
||||
assert result["results"][0]["status"] == "success"
|
||||
wb = load_workbook(result["results"][0]["file"])
|
||||
assert len(wb["Sheet1"]._images) == 1
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user