From 07f4d0fd5bec777db60b74112aebda79922aebde Mon Sep 17 00:00:00 2001 From: king Date: Tue, 15 Sep 2026 18:15:25 +0900 Subject: [PATCH] =?UTF-8?q?feat(project):=203=EB=8B=A8=EA=B3=84=20?= =?UTF-8?q?=EA=B5=AC=EC=A1=B0=20=EB=B0=B1=EC=97=94=EB=93=9C=20=EA=B8=B0?= =?UTF-8?q?=EB=B0=98=20=E2=80=94=20=EC=84=B8=EC=85=98=20=EC=82=AD=EC=A0=9C?= =?UTF-8?q?=EC=9E=90=C2=B7=ED=95=98=EC=9C=84=EC=97=85=EB=AC=B4=C2=B7?= =?UTF-8?q?=EB=A9=80=ED=8B=B0=ED=98=B8=EB=B0=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 프로젝트→세션→업무 3단계(아사나식) 전환의 1단계(백엔드). "세션"은 새 개념이 아니라 기존 "단계"(project_stages, 칸반 컬럼)를 그대로 재사용한다. - project_stages 에 created_by 추가 — 세션 삭제도 "만든 사람만"(_can_delete 재사용) 규칙 적용. api_delete_stage 를 _require_manage 단독 체크에서 생성자 체크로 변경. - 프로젝트 생성을 관리자 전용 → 로그인 사용자 전체로 개방 (api_create_project: _require_admin → _require_user). 홈 화면의 "새 프로젝트"/수정 모달·버튼도 is_admin 게이트를 걷어냄(생성자가 자동으로 owner 가 되어 _can_manage 통과 — 기존 로직 재사용). 멤버 배정 API는 계속 관리자 전용이라, 비관리자가 새 프로젝트 모달을 열면 멤버 선택란은 안내 문구만 보여주고 시도하지 않는다(불필요한 403 방지). - 하위 업무 — 새 테이블 없이 tasks.parent_task_id 재사용(부모 삭제 시 CASCADE). list_tasks 는 기본적으로 최상위만 반환(보드/캘린더/리스트에 하위업무가 카드로 안 뜨게) + comment_count 와 같은 방식으로 subtask_total/ subtask_done 집계 추가. create_task 에 parent_task_id 지원(하위업무는 부모와 같은 프로젝트, 세션 없음). - 멀티호밍(task_project_links, 아사나 기능) — 업무 하나가 여러 프로젝트/세션에 동시에 속할 수 있다. list_tasks(project_id=X, 비재귀)가 자동으로 그 프로젝트에 연결된 업무까지 포함하도록 확장(다른 호출 경로는 안 건드림 — 홈/내업무 전체보기에서 중복 노출 방지). 신규 API: GET /api/tasks/{id}(하위 업무·연결 포함 상세), POST/DELETE /api/tasks/{id}/links. - 설명란 이미지 삽입 준비: store.sanitize_description_html — 허용 태그만 남기고 스크립트/이벤트속성/위험 URL 스킴 제거(표준 html.parser, 외부 라이브러리 없음). api_create_task/api_update_task 저장 직전 통과. ⚠️ 배포 전 필수: scripts/sql/project_db_003_sections_subtasks_links.sql 를 project_db 에 적용해야 한다(project_stages.created_by, tasks.parent_task_id, task_project_links 없이는 이 커밋의 쿼리가 전부 실패한다). 그 안의 기존 설명 텍스트→HTML 이스케이프 UPDATE 문은 1회만 실행할 것. 프론트엔드(세션 드래그 재정렬, 사이드바 프로젝트별 보기, 하위업무/멀티호밍 UI, 설명란 이미지 붙여넣기 편집기)는 다음 커밋에서 이어간다. Co-Authored-By: Claude Sonnet 5 --- app/modules/project/db.py | 194 ++++++++++++++++-- app/modules/project/router.py | 81 +++++++- app/modules/project/store.py | 117 +++++++++++ .../project/templates/project/index.html | 12 +- .../project/templates/project/project.html | 1 + app/static/project.js | 28 ++- ...project_db_003_sections_subtasks_links.sql | 59 ++++++ 7 files changed, 447 insertions(+), 45 deletions(-) create mode 100644 scripts/sql/project_db_003_sections_subtasks_links.sql diff --git a/app/modules/project/db.py b/app/modules/project/db.py index 5f09203..97f86ab 100644 --- a/app/modules/project/db.py +++ b/app/modules/project/db.py @@ -125,9 +125,9 @@ class ProjectStore: for i, (stage_name, is_done) in enumerate(store.DEFAULT_STAGES): conn.execute( "INSERT INTO project_stages " - "(project_id, name, sort_order, is_done_stage) " - "VALUES (%s,%s,%s,%s)", - (pid, stage_name, i, is_done), + "(project_id, name, sort_order, is_done_stage, created_by) " + "VALUES (%s,%s,%s,%s,%s)", + (pid, stage_name, i, is_done, store.norm_str(created_by, lower=True)), ) self._log(conn, project_id=pid, actor=created_by, action=store.ACTION_CREATED, detail=nm) @@ -238,11 +238,11 @@ class ProjectStore: return [self._serialize(r) for r in rows] def add_stage( - self, *, project_id: int, name: str, is_done_stage: bool = False + self, *, project_id: int, name: str, is_done_stage: bool = False, created_by: str = "" ) -> dict[str, Any]: nm = store.norm_str(name) if not nm: - raise ValueError("단계 이름은 필수입니다.") + raise ValueError("세션 이름은 필수입니다.") with self._pool.connection() as conn: nxt = conn.execute( "SELECT COALESCE(MAX(sort_order), -1) + 1 AS n " @@ -250,12 +250,20 @@ class ProjectStore: (project_id,), ).fetchone() row = conn.execute( - "INSERT INTO project_stages (project_id, name, sort_order, is_done_stage) " - "VALUES (%s,%s,%s,%s) RETURNING *", - (project_id, nm, int(nxt["n"]), bool(is_done_stage)), + "INSERT INTO project_stages (project_id, name, sort_order, is_done_stage, created_by) " + "VALUES (%s,%s,%s,%s,%s) RETURNING *", + (project_id, nm, int(nxt["n"]), bool(is_done_stage), + store.norm_str(created_by, lower=True)), ).fetchone() return self._serialize(row) + def get_stage(self, *, stage_id: int) -> dict[str, Any] | None: + with self._pool.connection() as conn: + row = conn.execute( + "SELECT * FROM project_stages WHERE id = %s", (stage_id,) + ).fetchone() + return self._serialize(row) if row else None + def delete_stage(self, *, stage_id: int) -> None: with self._pool.connection() as conn: cur = conn.execute("DELETE FROM project_stages WHERE id = %s", (stage_id,)) @@ -265,14 +273,47 @@ class ProjectStore: # ════════════════════════════════════════════════════════════ # 업무 (tasks) # ════════════════════════════════════════════════════════════ + # 업무 목록에 함께 붙이는 하위업무 집계(댓글수와 같은 방식). 여러 쿼리에서 + # 재사용하므로 상수로 뽑아둔다. + _SUBTASK_COUNTS_SQL = ( + "(SELECT COUNT(*) FROM tasks st WHERE st.parent_task_id = t.id) AS subtask_total, " + "(SELECT COUNT(*) FROM tasks st WHERE st.parent_task_id = t.id " + " AND st.completed_at IS NOT NULL) AS subtask_done" + ) + def list_tasks( self, *, project_id: int | None = None, assignee_email: str | None = None, include_subprojects: bool = False, + parent_task_id: int | None = None, + top_level_only: bool = True, + include_links: bool = True, limit: int = 1000, ) -> list[dict[str, Any]]: + """업무 목록. + + 기본은 최상위 업무만(`parent_task_id IS NULL`) — 하위 업무는 부모 안에서만 + 보인다(아사나와 동일, 보드/캘린더/리스트 카드로 따로 뜨지 않는다). + `parent_task_id` 를 주면 그 부모의 하위 업무만 받는다. + + `project_id` 하나만 지정하고(`include_subprojects=False`, 최상위 조회일 + 때) `include_links=True`(기본값)이면, 그 프로젝트에 멀티호밍 + (`task_project_links`)으로 연결된 다른 프로젝트의 업무도 함께 반환한다. + 프로젝트 상세 화면이 '이 프로젝트만' 보여줄 때 쓰는 경로가 이 조합이다. + 다른 조합(홈 전체보기·내 업무 등, `project_id=None`)은 연결 업무를 섞지 + 않는다 — 같은 업무가 여러 곳에서 중복으로 집계되는 것을 막기 위함이다. + """ + link_mode = ( + include_links and project_id is not None + and not include_subprojects and parent_task_id is None + ) + if link_mode: + return self._list_tasks_for_project_with_links( + project_id=project_id, assignee_email=assignee_email, limit=limit, + ) + clauses: list[str] = [] params: list[Any] = [] if project_id is not None: @@ -292,13 +333,19 @@ class ProjectStore: if assignee_email: clauses.append("t.assignee_email = %s") params.append(store.norm_str(assignee_email, lower=True)) + if parent_task_id is not None: + clauses.append("t.parent_task_id = %s") + params.append(parent_task_id) + elif top_level_only: + clauses.append("t.parent_task_id IS NULL") where = ("WHERE " + " AND ".join(clauses)) if clauses else "" params.append(int(limit)) with self._pool.connection() as conn: rows = conn.execute( f"SELECT t.*, p.name AS project_name, p.color AS project_color, " f" s.name AS stage_name, s.is_done_stage, " - f" (SELECT COUNT(*) FROM task_comments c WHERE c.task_id = t.id) AS comment_count " + f" (SELECT COUNT(*) FROM task_comments c WHERE c.task_id = t.id) AS comment_count, " + f" {self._SUBTASK_COUNTS_SQL} " f"FROM tasks t " f"JOIN projects p ON p.id = t.project_id " f"LEFT JOIN project_stages s ON s.id = t.stage_id " @@ -308,15 +355,61 @@ class ProjectStore: ).fetchall() return [self._serialize(r) for r in rows] + def _list_tasks_for_project_with_links( + self, *, project_id: int, assignee_email: str | None, limit: int, + ) -> list[dict[str, Any]]: + """`list_tasks` 의 "이 프로젝트 + 여기 연결된 업무" 경로. + + `t.*` 뒤에 같은 이름(`stage_id`)을 다시 셀렉트해 덮어쓴다 — psycopg 의 + dict_row 는 중복 컬럼명이 있으면 마지막 값으로 덮어쓰므로, "이 프로젝트 + 맥락에서의 세션"(연결이면 link.stage_id, 기본홈이면 t.stage_id)이 + 최종값이 된다. 원래 소속 프로젝트 배지(`project_name`/`project_color`)는 + `t.project_id` 기준 그대로 둬 "다른 프로젝트에서 온 업무"임을 알 수 있게 + 한다. + """ + clauses = [ + "t.parent_task_id IS NULL", + "(t.project_id = %(pid)s OR link.project_id = %(pid)s)", + ] + params: dict[str, Any] = {"pid": project_id, "limit": int(limit)} + if assignee_email: + clauses.append("t.assignee_email = %(assignee)s") + params["assignee"] = store.norm_str(assignee_email, lower=True) + where = "WHERE " + " AND ".join(clauses) + with self._pool.connection() as conn: + rows = conn.execute( + f""" + SELECT t.*, p.name AS project_name, p.color AS project_color, + (SELECT COUNT(*) FROM task_comments c WHERE c.task_id = t.id) AS comment_count, + {self._SUBTASK_COUNTS_SQL}, + (link.id IS NOT NULL) AS is_linked, + COALESCE(es.name, s.name) AS stage_name, + COALESCE(es.is_done_stage, s.is_done_stage) AS is_done_stage, + COALESCE(link.stage_id, t.stage_id) AS stage_id + FROM tasks t + JOIN projects p ON p.id = t.project_id + LEFT JOIN project_stages s ON s.id = t.stage_id + LEFT JOIN task_project_links link + ON link.task_id = t.id AND link.project_id = %(pid)s + AND t.project_id <> %(pid)s + LEFT JOIN project_stages es ON es.id = link.stage_id + {where} + ORDER BY t.sort_order ASC, t.id ASC LIMIT %(limit)s + """, + params, + ).fetchall() + return [self._serialize(r) for r in rows] + def get_task(self, *, task_id: int) -> dict[str, Any] | None: with self._pool.connection() as conn: row = conn.execute( - "SELECT t.*, p.name AS project_name, p.color AS project_color, " - " s.name AS stage_name, s.is_done_stage, " - " (SELECT COUNT(*) FROM task_comments c WHERE c.task_id = t.id) AS comment_count " - "FROM tasks t JOIN projects p ON p.id = t.project_id " - "LEFT JOIN project_stages s ON s.id = t.stage_id " - "WHERE t.id = %s", + f"SELECT t.*, p.name AS project_name, p.color AS project_color, " + f" s.name AS stage_name, s.is_done_stage, " + f" (SELECT COUNT(*) FROM task_comments c WHERE c.task_id = t.id) AS comment_count, " + f" {self._SUBTASK_COUNTS_SQL} " + f"FROM tasks t JOIN projects p ON p.id = t.project_id " + f"LEFT JOIN project_stages s ON s.id = t.stage_id " + f"WHERE t.id = %s", (task_id,), ).fetchone() return self._serialize(row) if row else None @@ -336,6 +429,7 @@ class ProjectStore: start_time: str | None = None, due_time: str | None = None, created_by: str = "", + parent_task_id: int | None = None, ) -> dict[str, Any]: ttl = store.validate_task_title(title) pr = store.validate_priority(priority) @@ -346,8 +440,18 @@ class ProjectStore: assignee = store.norm_str(assignee_email, lower=True) with self._pool.connection() as conn: with conn.transaction(): - # stage 미지정 시 프로젝트의 첫 단계로 - if stage_id is None: + if parent_task_id is not None: + # 하위 업무는 부모와 같은 프로젝트에 속하고(담당자 선택지 등이 + # 맞아떨어지게), 보드에는 안 보이므로 세션을 갖지 않는다. + parent = conn.execute( + "SELECT project_id FROM tasks WHERE id = %s", (parent_task_id,) + ).fetchone() + if not parent: + raise KeyError(f"상위 업무를 찾을 수 없습니다: {parent_task_id}") + project_id = parent["project_id"] + stage_id = None + elif stage_id is None: + # stage 미지정 시 프로젝트의 첫 세션으로 st = conn.execute( "SELECT id FROM project_stages WHERE project_id = %s " "ORDER BY sort_order ASC, id ASC LIMIT 1", @@ -362,14 +466,14 @@ class ProjectStore: row = conn.execute( """ INSERT INTO tasks - (project_id, stage_id, title, description, assignee_email, - assignee_name, priority, start_date, due_date, + (project_id, stage_id, parent_task_id, title, description, + assignee_email, assignee_name, priority, start_date, due_date, start_time, due_time, sort_order, created_by) - VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) + VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) RETURNING * """, ( - project_id, stage_id, ttl, store.norm_str(description), + project_id, stage_id, parent_task_id, ttl, store.norm_str(description), assignee, store.norm_str(assignee_name), pr, sd, dd, sti, dti, int(nxt["n"]), store.norm_str(created_by, lower=True), ), @@ -482,6 +586,51 @@ class ProjectStore: ).fetchone() return bool(row and row["is_done_stage"]) + # ════════════════════════════════════════════════════════════ + # 멀티호밍 (task_project_links) — 업무 1개가 여러 프로젝트/세션에 동시 소속. + # 원래 소속(tasks.project_id/stage_id)은 여기서 건드리지 않는다("기본 홈"은 + # 업무 삭제로만 없어진다). 같은 프로젝트에 중복 연결하면 세션만 갱신한다. + # ════════════════════════════════════════════════════════════ + def add_task_link( + self, *, task_id: int, project_id: int, stage_id: int | None = None, + created_by: str = "", + ) -> dict[str, Any]: + with self._pool.connection() as conn: + row = conn.execute( + """ + INSERT INTO task_project_links (task_id, project_id, stage_id, created_by) + VALUES (%s,%s,%s,%s) + ON CONFLICT (task_id, project_id) DO UPDATE + SET stage_id = EXCLUDED.stage_id + RETURNING * + """, + (task_id, project_id, stage_id, store.norm_str(created_by, lower=True)), + ).fetchone() + return self._serialize(row) + + def remove_task_link(self, *, task_id: int, project_id: int) -> None: + with self._pool.connection() as conn: + conn.execute( + "DELETE FROM task_project_links WHERE task_id = %s AND project_id = %s", + (task_id, project_id), + ) + + def list_task_links(self, *, task_id: int) -> list[dict[str, Any]]: + with self._pool.connection() as conn: + rows = conn.execute( + """ + SELECT l.*, p.name AS project_name, p.color AS project_color, + s.name AS stage_name + FROM task_project_links l + JOIN projects p ON p.id = l.project_id + LEFT JOIN project_stages s ON s.id = l.stage_id + WHERE l.task_id = %s + ORDER BY l.created_at ASC + """, + (task_id,), + ).fetchall() + return [self._serialize(r) for r in rows] + # ════════════════════════════════════════════════════════════ # 활동 이력 # ════════════════════════════════════════════════════════════ @@ -748,7 +897,8 @@ class ProjectStore: out[k] = v.strftime("%H:%M") elif isinstance(v, date): out[k] = v.isoformat() - for k in ("id", "parent_id", "project_id", "stage_id", "task_id", "sort_order"): + for k in ("id", "parent_id", "project_id", "stage_id", "task_id", "sort_order", + "parent_task_id"): if k in out and out[k] is not None: try: out[k] = int(out[k]) diff --git a/app/modules/project/router.py b/app/modules/project/router.py index de0d47b..d593e38 100644 --- a/app/modules/project/router.py +++ b/app/modules/project/router.py @@ -570,6 +570,7 @@ async def project_page(request: Request, project_id: int) -> HTMLResponse: "page_title": project["name"], "page_subtitle": project.get("description") or "프로젝트 보드", "project": project, + "current_project_id": project_id, "nav_projects": nav_projects, "stages": stages, "tasks": all_tasks, @@ -619,8 +620,8 @@ async def api_list_projects(request: Request) -> JSONResponse: async def api_create_project( request: Request, payload: dict[str, Any] = Body(...) ) -> JSONResponse: - """최상위 프로젝트 생성 — 관리자 전용.""" - user = _require_admin(request) + """최상위 프로젝트 생성 — 로그인해 프로젝트 모듈에 접근할 수 있는 사용자 누구나.""" + user = _require_user(request) st = _db_or_503(request) try: project = st.create_project( @@ -778,6 +779,7 @@ async def api_add_stage( project_id=project_id, name=payload.get("name", ""), is_done_stage=bool(payload.get("is_done_stage", False)), + created_by=user["email"], ) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) @@ -788,13 +790,20 @@ async def api_add_stage( async def api_delete_stage( request: Request, project_id: int, stage_id: int ) -> JSONResponse: + """세션(단계) 삭제 — 만든 사람만(슈퍼관리자 예외). 생성자 정보가 없는 과거 + 데이터는 관리자에게 허용(잠김 방지) — `_can_delete` 규칙 그대로 재사용.""" user = _require_user(request) st = _db_or_503(request) _require_manage(request, st, user, project_id) + stage = st.get_stage(stage_id=stage_id) + if stage is None: + raise HTTPException(status_code=404, detail="세션을 찾을 수 없습니다.") + if not _can_delete(user, stage.get("created_by")): + raise HTTPException(status_code=403, detail=_DELETE_DENIED) try: st.delete_stage(stage_id=stage_id) except KeyError: - raise HTTPException(status_code=404, detail="단계를 찾을 수 없습니다.") + raise HTTPException(status_code=404, detail="세션을 찾을 수 없습니다.") return JSONResponse({"ok": True}) @@ -834,7 +843,7 @@ async def api_create_task( project_id=project_id, title=payload.get("title", ""), stage_id=payload.get("stage_id"), - description=payload.get("description", ""), + description=store.sanitize_description_html(payload.get("description", "")), assignee_email=payload.get("assignee_email", ""), assignee_name=payload.get("assignee_name", ""), priority=payload.get("priority", "normal"), @@ -843,9 +852,12 @@ async def api_create_task( start_time=payload.get("start_time"), due_time=payload.get("due_time"), created_by=user["email"], + parent_task_id=payload.get("parent_task_id"), ) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) + except KeyError as exc: + raise HTTPException(status_code=404, detail=str(exc)) # 생성과 동시에 담당자 지정 시 관리자 메일 + 담당자 인앱 알림 if task.get("assignee_email"): _notify( @@ -886,6 +898,8 @@ async def api_update_task( if existing is None: raise HTTPException(status_code=404, detail="업무를 찾을 수 없습니다.") _require_manage(request, st, user, existing["project_id"]) + if "description" in payload: + payload["description"] = store.sanitize_description_html(payload["description"]) try: task = st.update_task(task_id=task_id, fields=payload, actor=user["email"]) except ValueError as exc: @@ -952,6 +966,65 @@ async def api_delete_task(request: Request, task_id: int) -> JSONResponse: return JSONResponse({"ok": True}) +@router.get("/api/tasks/{task_id}") +async def api_get_task(request: Request, task_id: int) -> JSONResponse: + """업무 상세 — 하위 업무·연결된 프로젝트 포함. 업무 편집 팝업을 열 때 쓴다.""" + user = _require_user(request) + st = _db_or_503(request) + task = st.get_task(task_id=task_id) + if task is None: + raise HTTPException(status_code=404, detail="업무를 찾을 수 없습니다.") + _require_manage(request, st, user, task["project_id"]) + task["subtasks"] = st.list_tasks(parent_task_id=task_id) + task["links"] = st.list_task_links(task_id=task_id) + return JSONResponse({"task": task}) + + +# ──────────────────────────────────────────────────────────── +# JSON API — 업무 멀티호밍 (다른 프로젝트/세션에 연결) +# ──────────────────────────────────────────────────────────── +@router.post("/api/tasks/{task_id}/links") +async def api_add_task_link( + request: Request, task_id: int, payload: dict[str, Any] = Body(...) +) -> JSONResponse: + """업무를 다른 프로젝트(의 한 세션)에도 연결한다(아사나의 멀티호밍). + + 원래 소속(프로젝트)과 연결할 대상 프로젝트 양쪽에 대한 관리 권한이 필요하다. + """ + user = _require_user(request) + st = _db_or_503(request) + task = st.get_task(task_id=task_id) + if task is None: + raise HTTPException(status_code=404, detail="업무를 찾을 수 없습니다.") + _require_manage(request, st, user, task["project_id"]) + try: + target_project_id = int(payload.get("project_id")) + except (TypeError, ValueError): + raise HTTPException(status_code=400, detail="project_id 가 필요합니다.") + if target_project_id == task["project_id"]: + raise HTTPException(status_code=400, detail="이미 이 업무의 기본 프로젝트입니다.") + _require_manage(request, st, user, target_project_id) + stage_id = payload.get("stage_id") + link = st.add_task_link( + task_id=task_id, project_id=target_project_id, + stage_id=int(stage_id) if stage_id else None, created_by=user["email"], + ) + return JSONResponse({"link": link}, status_code=201) + + +@router.delete("/api/tasks/{task_id}/links/{project_id}") +async def api_remove_task_link(request: Request, task_id: int, project_id: int) -> JSONResponse: + user = _require_user(request) + st = _db_or_503(request) + task = st.get_task(task_id=task_id) + if task is None: + raise HTTPException(status_code=404, detail="업무를 찾을 수 없습니다.") + _require_manage(request, st, user, task["project_id"]) + _require_manage(request, st, user, project_id) + st.remove_task_link(task_id=task_id, project_id=project_id) + return JSONResponse({"ok": True}) + + @router.get("/api/projects/{project_id}/activity") async def api_activity(request: Request, project_id: int) -> JSONResponse: _require_user(request) diff --git a/app/modules/project/store.py b/app/modules/project/store.py index 7abf8b5..4274872 100644 --- a/app/modules/project/store.py +++ b/app/modules/project/store.py @@ -6,6 +6,7 @@ DB 없이 단위 테스트 가능한 순수 함수만 둔다(malaysia.store 패 from __future__ import annotations +from html.parser import HTMLParser from typing import Any # 프로젝트 생성 시 자동으로 깔리는 기본 진행 단계(칸반 컬럼). @@ -134,3 +135,119 @@ def build_subject_completed(*, project_name: str, task_title: str) -> str: def build_subject_project_done(*, project_name: str) -> str: return f"[DBX 프로젝트] 프로젝트 완료: {project_name}" + + +# ════════════════════════════════════════════════════════════ +# 업무 설명 HTML 정리 (이미지 붙여넣기/업로드 삽입 지원) +# +# 업무 편집창의 설명란은 contenteditable 이라 브라우저가 임의의 HTML을 +# 만들어 보낼 수 있다. 여기서 허용 태그만 남기고 스크립트/이벤트 속성/위험 +# URL 스킴을 제거한다 — 외부 라이브러리(bleach 등) 없이 표준 라이브러리 +# html.parser 만으로 구현한다(카페24 모듈 자체호스팅 원칙과 동일). +# +# 기존에 일반 텍스트로 저장돼 있던 설명은 마이그레이션(project_db_003)에서 +# 한 번에 안전한 HTML로 변환해뒀다 — 여기서는 "이게 HTML이냐 텍스트냐"를 +# 런타임에 추측하지 않는다(추측은 오탐이 난다). description 컬럼은 이제 +# 항상 HTML이라고 가정한다. +# ════════════════════════════════════════════════════════════ +_DESC_ALLOWED_TAGS: frozenset[str] = frozenset({ + "p", "br", "b", "strong", "i", "em", "u", "a", "img", + "ul", "ol", "li", "blockquote", "code", "pre", "span", "div", +}) +_DESC_VOID_TAGS: frozenset[str] = frozenset({"br", "img"}) +_DESC_ALLOWED_ATTRS: dict[str, frozenset[str]] = { + "a": frozenset({"href"}), + "img": frozenset({"src", "alt", "width", "height"}), +} +_DESC_DANGEROUS_URL_PREFIXES: tuple[str, ...] = ("javascript:", "vbscript:", "data:text/html") +# script/style 은 태그 자체는 물론 그 안의 텍스트(코드)도 통째로 버린다 — +# 허용 태그 목록에 없어 태그는 이미 안 남지만, HTMLParser는 이 둘의 내용을 +# 별도 데이터로 넘기므로 handle_data 에서 따로 걸러야 한다. +_DESC_RAW_TEXT_TAGS: frozenset[str] = frozenset({"script", "style"}) + + +def _desc_escape_text(text: str) -> str: + return text.replace("&", "&").replace("<", "<").replace(">", ">") + + +def _desc_escape_attr(value: str) -> str: + return _desc_escape_text(value).replace('"', """) + + +def _desc_url_is_safe(value: str) -> bool: + v = value.strip().lower() + return not any(v.startswith(prefix) for prefix in _DESC_DANGEROUS_URL_PREFIXES) + + +class _DescriptionSanitizer(HTMLParser): + """허용 태그/속성만 통과시키는 최소 HTML 정리기.""" + + def __init__(self) -> None: + super().__init__(convert_charrefs=True) + self.out: list[str] = [] + self._open: list[str] = [] + self._raw = False #