feat(malaysia): 일자별 엑셀을 상품(행) x 날짜(열) 양식으로

요청 양식에 맞춰 전치. A=상품명, B=자체상품코드, C~ 는 'MM월 DD일'.
기간 내 모든 날짜를 열로 유지(빈 날 포함), 등록된 낱개 아이템은 이동이
없어도 행으로 표시. 수량 0 은 빈칸. 헤더 갈색 배경 + 흰 굵은글씨,
C2 틀고정, 데이터 영역 얇은 테두리.

기간 생략 시 이력 최소~최대 날짜를 사용하고 2년 초과는 400 으로 막는다.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
2026-09-09 23:38:59 +09:00
parent bc6e2a3417
commit 32cc8299e3
2 changed files with 67 additions and 76 deletions
+1 -1
View File
@@ -92,7 +92,7 @@ JSON API:
| GET | `/malaysia/api/stocktakes/{id}/result` | 재고조사 최종 계산 | | GET | `/malaysia/api/stocktakes/{id}/result` | 재고조사 최종 계산 |
| POST | `/malaysia/stocktakes/{id}/rack/bulk` | 랙 입력 일괄 저장(병렬배열 `rk_cell/rk_sku/rk_upb/rk_box`) → 라인 재생성 | | POST | `/malaysia/stocktakes/{id}/rack/bulk` | 랙 입력 일괄 저장(병렬배열 `rk_cell/rk_sku/rk_upb/rk_box`) → 라인 재생성 |
| GET | `/malaysia/movements/export` | 특정 일자/종류의 낱개 수량 엑셀 | | GET | `/malaysia/movements/export` | 특정 일자/종류의 낱개 수량 엑셀 |
| GET | `/malaysia/movements/export-daily` | 기간 전체를 **일자 × 낱개 아이템** 수량으로 엑셀(`wh/type/from/to`). 콤보는 이미 구성 낱개 OUT 으로 저장돼 있어 자동 합산. 시트 `일자별`(피벗) + `상세`(긴 형식) | | GET | `/malaysia/movements/export-daily` | 기간 전체를 **행=낱개 상품 / 열=일자** 로 엑셀(`wh/type/from/to`). A=상품명, B=자체상품코드, C~=`MM월 DD일`. 콤보는 이미 구성 낱개 OUT 으로 저장돼 있어 자동 합산. 기간 생략 시 이력 최소~최대(최대 2년), 빈 날짜/무이동 상품도 줄·칸 유지 |
--- ---
+66 -75
View File
@@ -483,16 +483,18 @@ async def movements_export(request: Request) -> StreamingResponse:
@router.get("/movements/export-daily") @router.get("/movements/export-daily")
async def movements_export_daily(request: Request) -> StreamingResponse: async def movements_export_daily(request: Request) -> StreamingResponse:
"""기간 전체를 일자별 × 낱개 아이템 수량으로 엑셀 내보내기. """기간 전체를 상품(행) × 일자(열) 표로 엑셀 내보내기.
콤보(세트)는 등록 시 이미 BOM 으로 분해돼 구성 낱개 OUT movement 로 저장되므로 콤보(세트)는 등록 시 이미 BOM 으로 분해돼 구성 낱개 OUT movement 로 저장되므로
(`db.create_set_out`) 여기 합계는 "낱개 + 세트 분해분"이 모두 더해진 값이다. (`db.create_set_out`) 여기 수량은 "낱개 + 세트 분해분"이 모두 더해진 값이다.
쿼리: wh(창고), type(IN|OUT|ADJUST|STOCKTAKE, 기본 OUT), from, to(YYYY-MM-DD, 생략 시 전체) 쿼리: wh(창고), type(IN|OUT|ADJUST|STOCKTAKE, 기본 OUT), from, to(YYYY-MM-DD)
시트1 '일자별' — 행=날짜,=낱개 아이템(코드/이름 2줄 헤더), 마지막 열/행 합계. 기간 생략 시 이력의 최소~최대 날짜.은 기간 내 모든 날짜(빈 날 포함),
시트2 '상세' — 날짜/코드/사방넷코드/이름/수량 긴 형식. 행은 등록된 낱개 아이템 전체(수량 없는 아이템도 빈 줄로 표시).
A=상품명, B=자체상품코드, C~=`MM월 DD일`. 수량 0 은 빈칸.
""" """
import io # noqa: WPS433 import io # noqa: WPS433
from datetime import date as _date, timedelta as _timedelta # noqa: WPS433
guard = _guard(request) guard = _guard(request)
if not isinstance(guard, tuple): if not isinstance(guard, tuple):
@@ -510,95 +512,84 @@ async def movements_export_daily(request: Request) -> StreamingResponse:
date_from=dfrom or None, date_to=dto or None, date_from=dfrom or None, date_to=dto or None,
) )
reader = _reader(request) # {코드: {날짜: 수량}}
sabang = reader.sabangnet_code_map() if reader else {}
names = reader.name_map() if reader else {}
item_names = st.item_name_map()
def _name(code: str) -> str:
return names.get(code) or item_names.get(code) or code
# 열 순서: 등록된 낱개 아이템 순서 우선, 이력에만 있는 코드는 뒤에 붙인다.
used = {r["item_code"] for r in rows if r["qty"]}
codes = [it["item_code"] for it in st.list_items(kind="individual") if it["item_code"] in used]
codes += sorted(used - set(codes))
# {날짜: {코드: 수량}}
grid: dict[str, dict[str, int]] = {} grid: dict[str, dict[str, int]] = {}
for r in rows: for r in rows:
if not r["qty"]: if not r["qty"]:
continue continue
grid.setdefault(r["movement_date"], {})[r["item_code"]] = r["qty"] grid.setdefault(r["item_code"], {})[r["movement_date"]] = r["qty"]
dates = sorted(grid.keys())
# 열(날짜) — 지정 기간 우선, 없으면 이력 최소~최대. 빈 날짜도 열로 유지.
seen = sorted({d for per in grid.values() for d in per})
first = dfrom or (seen[0] if seen else today_kst().isoformat())
last = dto or (seen[-1] if seen else today_kst().isoformat())
try:
d0, d1 = _date.fromisoformat(first[:10]), _date.fromisoformat(last[:10])
except ValueError:
raise HTTPException(status_code=400, detail="날짜 형식은 YYYY-MM-DD 이어야 합니다.")
if d1 < d0:
raise HTTPException(status_code=400, detail="시작일이 종료일보다 늦습니다.")
if (d1 - d0).days > 732:
raise HTTPException(status_code=400, detail="기간이 너무 깁니다(최대 2년). 기간을 좁혀 주세요.")
dates = [(d0 + _timedelta(days=i)).isoformat() for i in range((d1 - d0).days + 1)]
# 행(상품) — 등록된 낱개 아이템 전체 + 이력에만 있는 코드는 뒤에 추가.
reader = _reader(request)
names = reader.name_map() if reader else {}
item_names = st.item_name_map()
items = st.list_items(kind="individual")
codes = [it["item_code"] for it in items]
codes += sorted(set(grid) - set(codes))
def _name(code: str) -> str:
return names.get(code) or item_names.get(code) or code
from openpyxl import Workbook # noqa: WPS433 from openpyxl import Workbook # noqa: WPS433
from openpyxl.styles import Alignment, Font, PatternFill # noqa: WPS433 from openpyxl.styles import Alignment, Border, Font, PatternFill, Side # noqa: WPS433
from openpyxl.utils import get_column_letter # noqa: WPS433 from openpyxl.utils import get_column_letter # noqa: WPS433
fill = PatternFill(start_color="FF4472C4", end_color="FF4472C4", fill_type="solid")
head_font = Font(color="FFFFFFFF", bold=True)
center = Alignment(horizontal="center", vertical="center", wrap_text=True)
wb = Workbook() wb = Workbook()
ws = wb.active ws = wb.active
ws.title = "일자별" ws.title = f"{mtype}_일자별"
# 2줄 헤더: 1행 코드, 2행 이름
ws.append(["날짜"] + codes + ["합계"])
ws.append([mtype] + [_name(c) for c in codes] + [""])
for d in dates:
line = grid[d]
vals = [int(line.get(c, 0)) for c in codes]
ws.append([d] + vals + [sum(vals)])
if dates:
totals = [sum(int(grid[d].get(c, 0)) for d in dates) for c in codes]
ws.append(["합계"] + totals + [sum(totals)])
last = ws.max_row
for col in range(1, len(codes) + 3):
ws.cell(row=last, column=col).font = Font(bold=True)
for col in range(1, len(codes) + 3): header = ["상품명", "자체상품코드"] + [f"{d[5:7]}{d[8:10]}" for d in dates]
for row in (1, 2): ws.append(header)
cell = ws.cell(row=row, column=col) for code in codes:
cell.fill = fill per = grid.get(code, {})
cell.font = head_font ws.append(
cell.alignment = center [_name(code), code]
ws.freeze_panes = "B3" + [(int(per[d]) if per.get(d) else None) for d in dates]
ws.column_dimensions["A"].width = 13 )
for col in range(2, len(codes) + 3):
ws.column_dimensions[get_column_letter(col)].width = 12
# 시트2 — 긴 형식(피벗/필터용) ncols = len(header)
ws2 = wb.create_sheet("상세") fill = PatternFill(start_color="FF9B3C28", end_color="FF9B3C28", fill_type="solid")
header2 = ["날짜", "아이템코드", "사방넷코드", "이름", "수량"] head_font = Font(color="FFFFFFFF", bold=True)
ws2.append(header2) center = Alignment(horizontal="center", vertical="center")
for d in dates: thin = Side(style="thin", color="FFBFBFBF")
for c in codes: box = Border(left=thin, right=thin, top=thin, bottom=thin)
qty = int(grid[d].get(c, 0)) for col in range(1, ncols + 1):
if qty == 0: cell = ws.cell(row=1, column=col)
continue
ws2.append([d, c, sabang.get(c, ""), _name(c), qty])
for col in range(1, len(header2) + 1):
cell = ws2.cell(row=1, column=col)
cell.fill = fill cell.fill = fill
cell.font = head_font cell.font = head_font
cell.alignment = center cell.alignment = center
ws2.freeze_panes = "A2" for row in range(2, ws.max_row + 1):
if ws2.max_row > 1: for col in range(1, ncols + 1):
ws2.auto_filter.ref = f"A1:{get_column_letter(len(header2))}{ws2.max_row}" cell = ws.cell(row=row, column=col)
for col in range(1, len(header2) + 1): cell.border = box
letter = get_column_letter(col) if col >= 3:
longest = 0 cell.alignment = center
for cell in ws2[letter]:
v = "" if cell.value is None else str(cell.value) ws.freeze_panes = "C2"
width = sum(2 if ord(ch) > 0x1100 else 1 for ch in v) longest = max((sum(2 if ord(ch) > 0x1100 else 1 for ch in _name(c)) for c in codes), default=8)
longest = max(longest, width) ws.column_dimensions["A"].width = max(14, min(40, longest + 2))
ws2.column_dimensions[letter].width = max(8, min(50, longest + 2)) ws.column_dimensions["B"].width = 16
for col in range(3, ncols + 1):
ws.column_dimensions[get_column_letter(col)].width = 11
buf = io.BytesIO() buf = io.BytesIO()
wb.save(buf) wb.save(buf)
buf.seek(0) buf.seek(0)
span = f"{dfrom or 'all'}_{dto or 'all'}" fname = f"malaysia_{mtype}_{d0.isoformat()}_{d1.isoformat()}.xlsx"
fname = f"malaysia_{mtype}_daily_{span}.xlsx"
return StreamingResponse( return StreamingResponse(
buf, buf,
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",