diff --git a/blog-lab/app/db.py b/blog-lab/app/db.py index e40bbce..0d45559 100644 --- a/blog-lab/app/db.py +++ b/blog-lab/app/db.py @@ -102,6 +102,22 @@ def init_db() -> None: ) """) + # 브랜드커넥트 제휴 링크 + conn.execute(""" + CREATE TABLE IF NOT EXISTS brand_links ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + post_id INTEGER REFERENCES blog_posts(id), + keyword_id INTEGER REFERENCES keyword_analyses(id), + url TEXT NOT NULL, + product_name TEXT NOT NULL DEFAULT '', + description TEXT NOT NULL DEFAULT '', + placement_hint TEXT NOT NULL DEFAULT '', + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')) + ) + """) + conn.execute("CREATE INDEX IF NOT EXISTS idx_bl_post ON brand_links(post_id)") + conn.execute("CREATE INDEX IF NOT EXISTS idx_bl_keyword ON brand_links(keyword_id)") + # 기본 프롬프트 템플릿 시딩 (존재하지 않을 때만) _seed_templates(conn) @@ -453,6 +469,94 @@ def delete_commission(comm_id: int) -> bool: return True +# ── brand_links CRUD ──────────────────────────────────────────────────────── + +def _bl_row_to_dict(r) -> Dict[str, Any]: + return { + "id": r["id"], + "post_id": r["post_id"], + "keyword_id": r["keyword_id"], + "url": r["url"], + "product_name": r["product_name"], + "description": r["description"], + "placement_hint": r["placement_hint"], + "created_at": r["created_at"], + } + + +def add_brand_link(data: Dict[str, Any]) -> Dict[str, Any]: + with _conn() as conn: + conn.execute( + """INSERT INTO brand_links (post_id, keyword_id, url, product_name, description, placement_hint) + VALUES (?, ?, ?, ?, ?, ?)""", + ( + data.get("post_id"), + data.get("keyword_id"), + data.get("url", ""), + data.get("product_name", ""), + data.get("description", ""), + data.get("placement_hint", ""), + ), + ) + row = conn.execute( + "SELECT * FROM brand_links WHERE rowid = last_insert_rowid()" + ).fetchone() + return _bl_row_to_dict(row) + + +def get_brand_links( + post_id: Optional[int] = None, + keyword_id: Optional[int] = None, +) -> List[Dict[str, Any]]: + with _conn() as conn: + if post_id is not None: + rows = conn.execute( + "SELECT * FROM brand_links WHERE post_id = ? ORDER BY id", (post_id,) + ).fetchall() + elif keyword_id is not None: + rows = conn.execute( + "SELECT * FROM brand_links WHERE keyword_id = ? ORDER BY id", (keyword_id,) + ).fetchall() + else: + rows = conn.execute("SELECT * FROM brand_links ORDER BY id DESC LIMIT 100").fetchall() + return [_bl_row_to_dict(r) for r in rows] + + +def update_brand_link(link_id: int, data: Dict[str, Any]) -> Optional[Dict[str, Any]]: + with _conn() as conn: + fields = [] + values = [] + for k in ("post_id", "keyword_id", "url", "product_name", "description", "placement_hint"): + if k in data: + fields.append(f"{k} = ?") + values.append(data[k]) + if not fields: + row = conn.execute("SELECT * FROM brand_links WHERE id = ?", (link_id,)).fetchone() + return _bl_row_to_dict(row) if row else None + values.append(link_id) + conn.execute(f"UPDATE brand_links SET {', '.join(fields)} WHERE id = ?", values) + row = conn.execute("SELECT * FROM brand_links WHERE id = ?", (link_id,)).fetchone() + return _bl_row_to_dict(row) if row else None + + +def delete_brand_link(link_id: int) -> bool: + with _conn() as conn: + row = conn.execute("SELECT id FROM brand_links WHERE id = ?", (link_id,)).fetchone() + if not row: + return False + conn.execute("DELETE FROM brand_links WHERE id = ?", (link_id,)) + return True + + +def link_brand_links_to_post(keyword_id: int, post_id: int) -> None: + """keyword_id로 등록된 링크들을 post_id에도 연결.""" + with _conn() as conn: + conn.execute( + "UPDATE brand_links SET post_id = ? WHERE keyword_id = ? AND post_id IS NULL", + (post_id, keyword_id), + ) + + def get_dashboard_stats() -> Dict[str, Any]: """대시보드 집계: 총 포스트/클릭/구매/수익 + 월별 추이.""" with _conn() as conn: diff --git a/blog-lab/tests/test_db_brand_links.py b/blog-lab/tests/test_db_brand_links.py new file mode 100644 index 0000000..a84d7b2 --- /dev/null +++ b/blog-lab/tests/test_db_brand_links.py @@ -0,0 +1,67 @@ +"""brand_links DB CRUD 테스트.""" +import os +import pytest +from app import db +from app.config import DB_PATH + + +@pytest.fixture(autouse=True) +def setup_db(tmp_path): + """테스트용 임시 DB 사용.""" + test_db = str(tmp_path / "test.db") + import app.config as config + config.DB_PATH = test_db + db.DB_PATH = test_db + db.init_db() + yield + + +def test_add_brand_link(): + link = db.add_brand_link({ + "keyword_id": 1, + "url": "https://link.coupang.com/abc", + "product_name": "테스트 상품", + "description": "상품 설명", + "placement_hint": "본문 중간", + }) + assert link["id"] is not None + assert link["url"] == "https://link.coupang.com/abc" + assert link["product_name"] == "테스트 상품" + assert link["keyword_id"] == 1 + assert link["post_id"] is None + + +def test_get_brand_links_by_keyword_id(): + db.add_brand_link({"keyword_id": 1, "url": "https://a.com", "product_name": "A"}) + db.add_brand_link({"keyword_id": 1, "url": "https://b.com", "product_name": "B"}) + db.add_brand_link({"keyword_id": 2, "url": "https://c.com", "product_name": "C"}) + links = db.get_brand_links(keyword_id=1) + assert len(links) == 2 + + +def test_get_brand_links_by_post_id(): + db.add_brand_link({"post_id": 10, "url": "https://a.com", "product_name": "A"}) + links = db.get_brand_links(post_id=10) + assert len(links) == 1 + assert links[0]["post_id"] == 10 + + +def test_update_brand_link(): + link = db.add_brand_link({"url": "https://a.com", "product_name": "원래 이름"}) + updated = db.update_brand_link(link["id"], {"product_name": "새 이름", "post_id": 5}) + assert updated["product_name"] == "새 이름" + assert updated["post_id"] == 5 + + +def test_delete_brand_link(): + link = db.add_brand_link({"url": "https://a.com", "product_name": "삭제할 링크"}) + assert db.delete_brand_link(link["id"]) is True + assert db.delete_brand_link(link["id"]) is False + + +def test_link_keyword_to_post(): + db.add_brand_link({"keyword_id": 1, "url": "https://a.com", "product_name": "A"}) + db.add_brand_link({"keyword_id": 1, "url": "https://b.com", "product_name": "B"}) + db.link_brand_links_to_post(keyword_id=1, post_id=10) + links = db.get_brand_links(post_id=10) + assert len(links) == 2