feat(blog-lab): brand_links 테이블 및 CRUD 추가

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

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:

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