【機能実装編】Flaskアプリにレコメンド機能を追加する

1. 概要

• この記事でできること
  Flaskアプリに「おすすめ表示（レコメンド機能）」を追加します。
  • 外部APIから人気データを取得
  • 取得データをアプリ内部の形式に変換（正規化）
  • SQLiteに保存して再利用可能に
  • トップページ / にカードUIとして一覧表示
• 事前に決めておくこと
  • なんのAPIを使うかを必ず決めてください。
  • 選んだAPIによって、エンドポイント・認証方式・レスポンス形式が異なるため、記事内の ★EDIT 部分をあなたのAPI仕様に合わせて変更します。

2. 依存ライブラリの準備 & フォルダ/ファイルを一括作成

Windows (PowerShell) の人はこちら

1. 必要ライブラリを requirements.txt に追加

コマンドプロンプト

"Flask-SQLAlchemy`nFlask-Migrate`nrequests" | Out-File -Append -Encoding utf8 requirements.txt

2. 一括インストール

コマンドプロンプト

pip install -r requirements.txt

3. 必要なフォルダを作成

コマンドプロンプト

mkdir app/services/providers, scripts

4. 必要なファイルを作成

コマンドプロンプト

ni app/models/catalog.py, app/services/recommend.py, app/services/providers/__init__.py, app/services/providers/base.py, scripts/ingest_catalog.py

Mac / Linux (bash / zsh) の人はこちら

1. 必要ライブラリを requirements.txt に追加

コマンドプロンプト

echo -e "Flask-SQLAlchemy\nFlask-Migrate\nrequests" >> requirements.txt

2. 一括インストール

コマンドプロンプト

pip install -r requirements.txt

3. 必要なフォルダを作成

コマンドプロンプト

mkdir -p app/services/providers scripts

4. 必要なファイルを作成

コマンドプロンプト

touch app/models/catalog.py app/services/recommend.py app/services/providers/__init__.py app/services/providers/base.py scripts/ingest_catalog.py

ライブラリの説明

• Flask-SQLAlchemy：
  • Flask に データベース機能 を追加する拡張。
  • SQL を直接書かずに、Pythonクラス（モデル）を使ってテーブルを定義・操作できる。
• Flask-Migrate：
  • データベースの マイグレーション管理ツール。
  • モデルの変更（カラム追加・削除、テーブル新規作成など）を安全に DB に反映できる。
  • Alembic を裏で使っており、コマンド操作でスキーマ変更をバージョン管理できる。
• requests：
  • 外部 API と通信するための HTTP クライアント。
  • GET / POST リクエストを簡単に書けるので、API からデータを取得・送信する際に便利。

3. 環境変数（.env）

.env

# Flask
SECRET_KEY=devkey

# DB（開発はSQLite）
USE_DB=true
DATABASE_URL=sqlite:///app.db

# ===== Provider 基本設定（必須）=====
PROVIDER=my_api
PROVIDER_BASE_URL=<YOUR_BASE_URL>

# ===== 認証（必要な方式だけ使う。不要ならコメントのまま）=====
# PROVIDER_API_KEY=xxxxx
# PROVIDER_AUTH_HEADER=Authorization
# PROVIDER_AUTH_PREFIX=Bearer
# PROVIDER_QUERY_KEY=api_key

# ===== 取得エンドポイント / ページング / レスポンス配列キー =====
# 人気/注目/新着など、実際に叩くパス
# PROVIDER_POPULAR_PATH=/movie/popular

# ページング（APIに合わせて）
# PROVIDER_PAGE_PARAM=page
# PROVIDER_LIMIT_PARAM=limit
# PROVIDER_DEFAULT_LIMIT=20

# レスポンスの配列キーが特殊な場合に明示（例: results, items, data, records など）
# PROVIDER_RESULTS_KEY=results

# ===== 画像パスの前置（相対パス→絶対URLに）=====
# PROVIDER_IMAGE_BASE=<YOUR_IMAGE_BASE>

# ===== 追加クエリ/ヘッダー（必要な場合のみ）=====
# PROVIDER_EXTRA_QUERY=language=ja|region=JP
# PROVIDER_EXTRA_HEADERS=Accept: application/json

# ===== タイムアウト =====
# Provider内部で使う秒数。なければデフォルト10
# PROVIDER_TIMEOUT=10

# （アプリ全体で別用途に使う場合は残す）
REQUEST_TIMEOUT=10         # 秒。APIが遅い時は15〜20に（※任意・アプリ側用）

# ===== アプリ側の表示・取り込み制御 =====
INGEST_PAGES=2             # 取り込みページ数（fetch_popular のページ数上限）
DEFAULT_TOP_K=9            # トップ画面で表示する件数

ファイルの説明

• .env にはアプリ全体の設定（Flask・DB・APIなど）を集約しています。
• APIキーやヘッダー名が未設定なら、コードは自動的に「認証なし」で動作します。
• 設定があれば、自動的に ヘッダー認証（Authorization: Bearer <KEY>）または クエリ認証（?api_key=<KEY>）を付与します。

コードの解説

• SECRET_KEY / DATABASE_URL
  Flaskの基本設定。開発ではSQLiteを使い、必要に応じて本番用に差し替え可能。
• PROVIDER / PROVIDER_BASE_URL
  利用するAPIの識別子とベースURL。
• PROVIDER_API_KEY / PROVIDER_AUTH_HEADER / PROVIDER_AUTH_PREFIX / PROVIDER_QUERY_KEY
  認証方式を切り替えるためのセット。
• PROVIDER_POPULAR_PATH / PROVIDER_PAGE_PARAM / PROVIDER_RESULTS_KEY
  人気な情報を取得する際のパスやページング方法、レスポンス配列キー。
• PROVIDER_IMAGE_BASE
  画像パス（poster_pathなど）に付与するベースURL。
• PROVIDER_EXTRA_QUERY / PROVIDER_EXTRA_HEADERS
  言語や地域を固定したい場合に利用。基本は不要なのでコメントアウトで残す。
• PROVIDER_TIMEOUT / REQUEST_TIMEOUT
  API呼び出し時のタイムアウト秒数。REQUEST_TIMEOUT はアプリ側での制御用。
• INGEST_PAGES / DEFAULT_TOP_K
  アプリの表示件数や取り込みページ数を制御。教材ではトップ画面のカードUIに反映される。

ここでやること！

あなたが使いたい API に応じて、どの変数を使うか／使わないか を判断する必要があります。

以下のプロンプトを生成AIにコピペしてください。

使用API：の後ろに、自分のコンセプトに合った API 名を1行で書くだけです。

あなたは Flask 教材のアシスタントです。以下の制約で回答してください。

## 入力
使用API:(例：ChatGPTのAPIを使用)

## あなたがやること
1) 公式ドキュメントから、そのAPIの認証方式を推論してください（以下のいずれか・複数可）。
   - 認証不要
   - ヘッダー認証（例: Authorization: Bearer <KEY>）
   - クエリ認証（例: ?api_key=<KEY>）

2) .env に必要な変数だけを抽出してください（太字で書く）。
   - 必須候補: PROVIDER_BASE_URL
   - 認証が必要な場合のみ: PROVIDER_API_KEY, PROVIDER_AUTH_HEADER, PROVIDER_AUTH_PREFIX, PROVIDER_QUERY_KEY
   - 必要に応じて（API仕様次第で）: 
     PROVIDER_POPULAR_PATH, PROVIDER_PAGE_PARAM, PROVIDER_LIMIT_PARAM, PROVIDER_DEFAULT_LIMIT,
     PROVIDER_RESULTS_KEY, PROVIDER_IMAGE_BASE, PROVIDER_EXTRA_QUERY, PROVIDER_EXTRA_HEADERS,
     PROVIDER_TIMEOUT

3) 使わないと判断できる変数は「使用しない変数」に**変数名だけ**列挙してください（理由を一言で）。

4) **完成コードは絶対に出さないこと**。  
   値の例は出して構いませんが、.env の各行は **太字** で表現してください（例：**PROVIDER_BASE_URL=https://...**）。

5) コメントアウト運用も説明してください。
   - 「使用する変数」は **コメントアウトを外して記入**する。
   - 「使用しない変数」は `.env` にコメントのまま残す（後で使いたくなったら解除できる）。
   - 条件次第で使うものは「必要ならコメント解除」と明記する。

## 出力フォーマット（厳守）

### 使用する変数
・**変数名=値**  
補足（なぜ必要か、どこから取得するか）

### 使用しない変数
・**変数名**  
補足（理由）

4. アプリ工場（app/__init__.py）

app/__init__.py

# このファイルはFlaskアプリの工場。設定・DB・Blueprintの登録をまとめて行う。
from flask import Flask
from .config import Config
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate

db = SQLAlchemy()
migrate = Migrate()

def create_app():
    app = Flask(__name__)
    app.config.from_object(Config)

    if app.config.get("USE_DB", True):
        app.config["SQLALCHEMY_DATABASE_URI"] = app.config.get("DATABASE_URL", "sqlite:///app.db")
        app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
        db.init_app(app)
        migrate.init_app(app, db)
        from . import models  # Alembicがモデルを検出できるように（超重要）

    from .routes.web import web_bp
    from .routes.api import api_bp
    app.register_blueprint(web_bp)                 # 画面ルート
    app.register_blueprint(api_bp, url_prefix="/api")  # /api 配下にAPIを集約
    return app

__all__ = ["create_app", "db"]

ファイルの説明

Flaskアプリ生成、DB初期化、Blueprint登録を行う工場関数。

コードの解説

• db = SQLAlchemy() / migrate = Migrate()： 拡張のインスタンス（後で init_app）。
• app.config.from_object(Config)： .env の値が反映された設定を読み込み。
• db.init_app(app) / migrate.init_app(app, db)： アプリにDB/マイグレ連携。
• from . import models： migrateが差分検出できるようにモデルを読み込む。
• app.register_blueprint(web_bp)： 画面ルートを有効化。
• app.register_blueprint(api_bp, url_prefix="/api")： APIルートを /api/... へ集約。

自分用に変更する箇所

• 基本なし。Blueprintを追加したい時はここに登録

5. 設定クラス（app/config.py）

app/config.py

# このファイルは.envの設定値をアプリ全体に渡す設定クラスを定義する。
import os
from dotenv import load_dotenv
load_dotenv()

class Config:
    SECRET_KEY = os.getenv("SECRET_KEY", "devkey")

    # DB
    USE_DB = os.getenv("USE_DB", "true").lower() == "true"
    DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///app.db")

    # Provider（共通）
    PROVIDER = os.getenv("PROVIDER", "")
    PROVIDER_BASE_URL = os.getenv("PROVIDER_BASE_URL", "")

    # 認証（オプション）
    PROVIDER_API_KEY = os.getenv("PROVIDER_API_KEY", "")
    PROVIDER_AUTH_HEADER = os.getenv("PROVIDER_AUTH_HEADER", "")
    PROVIDER_AUTH_PREFIX = os.getenv("PROVIDER_AUTH_PREFIX", "")
    PROVIDER_QUERY_KEY = os.getenv("PROVIDER_QUERY_KEY", "")

    # エンドポイント / ページング / 配列キー
    PROVIDER_POPULAR_PATH = os.getenv("PROVIDER_POPULAR_PATH", "")     # 例: /movie/popular
    PROVIDER_PAGE_PARAM = os.getenv("PROVIDER_PAGE_PARAM", "")         # 例: page
    PROVIDER_LIMIT_PARAM = os.getenv("PROVIDER_LIMIT_PARAM", "")       # 例: limit / per_page
    PROVIDER_DEFAULT_LIMIT = int(os.getenv("PROVIDER_DEFAULT_LIMIT", "0") or 0)
    PROVIDER_RESULTS_KEY = os.getenv("PROVIDER_RESULTS_KEY", "")       # 例: results / items

    # 画像の前置URL
    PROVIDER_IMAGE_BASE = os.getenv("PROVIDER_IMAGE_BASE", "")         # 例: https://image.tmdb.org/t/p/w500

    # 追加クエリ / 追加ヘッダー
    PROVIDER_EXTRA_QUERY = os.getenv("PROVIDER_EXTRA_QUERY", "")       # 例: language=ja|region=JP
    PROVIDER_EXTRA_HEADERS = os.getenv("PROVIDER_EXTRA_HEADERS", "")   # 例: Accept: application/json; X-Client: myapp

    # Provider専用タイムアウト（未設定ならデフォルト10）
    PROVIDER_TIMEOUT = int(os.getenv("PROVIDER_TIMEOUT", "10") or 10)

    # アプリ共通
    REQUEST_TIMEOUT = int(os.getenv("REQUEST_TIMEOUT", "10"))
    DEFAULT_TOP_K   = int(os.getenv("DEFAULT_TOP_K", "9"))
    INGEST_PAGES    = int(os.getenv("INGEST_PAGES", "2"))

ファイルの説明

.env の値をアプリ設定として公開。空であれば認証は無効の挙動にします。

コードの解説

• load_dotenv()： .env 読み込み。
• os.getenv()： 環境変数取得。
• PROVIDER_*： 外部API切替・URL・キー。
• PROVIDER_AUTH_* と PROVIDER_QUERY_KEY は空文字なら無視されます。

自分用に変更する箇所

• なし（値は .env で変更）

6. モデル（app/models/catalog.py, init.py）

app/models/catalog.py

# このファイルはCatalogモデル（DBテーブル定義）を管理する。
from datetime import datetime
from app import db

class Catalog(db.Model):
    id = db.Column(db.Integer, primary_key=True)                   # 自動採番PK
    provider = db.Column(db.String(32), nullable=False, index=True)  # どのAPI由来か
    external_id = db.Column(db.String(128), nullable=False, unique=True)  # 外部APIのID（重複禁止）

    title = db.Column(db.String(255), nullable=False)  # 作品名（必須）
    genre = db.Column(db.String(64))                   # 簡易ジャンル（任意）
    image_url = db.Column(db.String(512))              # 画像の完全URL
    summary = db.Column(db.Text)                       # 概要（長文可）
    release_date = db.Column(db.String(20))            # 公開日（文字列で簡易保持）
    popularity = db.Column(db.Float)                   # 人気スコア（APIの指標を流用）

    created_at = db.Column(db.DateTime, default=datetime.utcnow, index=True)  # 追加日時
    updated_at = db.Column(db.DateTime, default=datetime.utcnow,
                           onupdate=datetime.utcnow, index=True)              # 更新日時

app/models/__init__.py

from .catalog import Catalog
__all__ = ["Catalog"]

ファイルの説明

正規化後データを保存するDBテーブル（ORM定義）。

コードの解説

• 例）title = db.Column(db.String(255), nullable=False)
  • db.Column(<型>, <制約...>)：形式
  • db.String(255)：最大255文字の文字列
  • nullable=False：必須（NULL禁止）
• unique=True：一意制約
• index=True：索引作成。

自分用に変更する箇所

• なし

7. プロバイダ共通I/F（app/services/providers/base.py）

app/services/providers/base.py

# このファイルは外部APIとの共通インタフェースを定義する（ProviderBase）。
from typing import List, Dict, Any

# 正規化後の共通レコード形
# keys: external_id, title, genre, image_url, summary, release_date, popularity
NormalizedItem = Dict[str, Any]

class ProviderBase:
    def fetch_popular(self, page: int = 1) -> List[NormalizedItem]:
        """人気/注目/新着 等の一覧を返す。"""
        raise NotImplementedError

ファイルの説明

外部API実装の共通インタフェースと内部共通形。

コードの解説

• すべてのプロバイダが fetch_popular() を実装。
• 上位層はこの共通形だけ見れば動く。

自分用に変更する箇所

• なし

8. プロバイダの動的ロード（app/services/providers/__init__.py）

app/services/providers/__init__.py

# このファイルは.envのPROVIDER名から動的にプロバイダをロードする。
from importlib import import_module
from flask import current_app
from .base import ProviderBase

def get_provider() -> ProviderBase:
    """
    .env の PROVIDER 名をモジュールとして動的 import し、その中の `Provider` クラスをインスタンス化します。
    例: PROVIDER=foo → app/services/providers/foo.py に class Provider(ProviderBase) を用意
    """
    name = (current_app.config.get("PROVIDER") or "").strip()
    if not name:
        raise RuntimeError("PROVIDER is not set. Please set PROVIDER=<your_provider> in .env")

    try:
        module = import_module(f".{name}", __name__)        # app.services.providers.<name>
    except ModuleNotFoundError as e:
        raise RuntimeError(f"Provider module '{name}' not found under app/services/providers/") from e

    if not hasattr(module, "Provider"):
        raise RuntimeError(f"Provider module '{name}' must define a class named 'Provider'")

    provider = module.Provider()  # must implement ProviderBase
    if not isinstance(provider, ProviderBase):
        raise RuntimeError(f"'Provider' in '{name}' must inherit ProviderBase")
    return provider

ファイルの説明

.env の PROVIDER 名をそのままモジュール名としてロード。
app/services/providers/my_api.py に class Provider(ProviderBase) を置けば即認識。

コードの解説

• import_module(f".{name}", __name__)： providers.<name> を相対 import
• Provider クラスの存在・型をチェック → そのインスタンスを返す

自分用に変更する箇所

• なし

9. あなたのAPI実装テンプレ（app/services/providers/my_api.py）

コマンドプロンプト

ni app/services/providers/my_api.py

app/services/providers/my_api.py

# このファイルはあなたのAPI依存処理（通信・認証・正規化）を1ファイルに集約する。
import requests
from typing import List, Dict, Any, Union, Optional
from flask import current_app
from .base import ProviderBase, NormalizedItem

_S = requests.Session()

def _build_headers() -> Dict[str, str]:
    """ヘッダー認証（任意）。.envに設定があれば自動付与。"""
    header_name = current_app.config.get("PROVIDER_AUTH_HEADER") or ""
    prefix = current_app.config.get("PROVIDER_AUTH_PREFIX", "") or ""
    api_key = current_app.config.get("PROVIDER_API_KEY", "") or ""
    if header_name and api_key:
        return {header_name: f"{prefix} {api_key}".strip()}
    return {}

def _maybe_add_query_auth(params: Dict[str, Any]) -> None:
    """クエリ認証（任意）。.envに設定があれば自動付与。"""
    key_name = current_app.config.get("PROVIDER_QUERY_KEY") or ""
    api_key = current_app.config.get("PROVIDER_API_KEY", "") or ""
    if key_name and api_key:
        params.setdefault(key_name, api_key)

def _req(path: str, **params) -> Union[Dict[str, Any], List[Dict[str, Any]]]:
    base = current_app.config["PROVIDER_BASE_URL"].rstrip("/")
    _maybe_add_query_auth(params)
    headers = _build_headers()
    r = _S.get(f"{base}{path}", params=params, headers=headers, timeout=10)
    r.raise_for_status()
    return r.json()

_ARRAY_KEYS = ("results", "items", "data", "records", "list", "docs")

def _pick_list(data: Union[Dict[str, Any], List[Dict[str, Any]]]) -> List[Dict[str, Any]]:
    """配列キーの自動検出＋明示キー（PROVIDER_RESULTS_KEY）優先。"""
    if isinstance(data, list):
        return data
    if not isinstance(data, dict):
        return []
    forced = (current_app.config.get("PROVIDER_RESULTS_KEY") or "").strip()
    if forced and isinstance(data.get(forced), list):
        return data[forced]
    for k in _ARRAY_KEYS:
        v = data.get(k)
        if isinstance(v, list):
            return v
    # 1件のみ返すAPIに一応対応
    return [data]

def _img(url: Optional[str]) -> Optional[str]:
    """相対パス系の画像にベースURLを付与（PROVIDER_IMAGE_BASE）。"""
    if not url:
        return None
    s = str(url)
    if s.startswith("http://") or s.startswith("https://"):
        return s
    base = (current_app.config.get("PROVIDER_IMAGE_BASE") or "").rstrip("/")
    if not base:
        return s
    if not s.startswith("/"):
        s = "/" + s
    return f"{base}{s}"

def _to_float(v: Any) -> float:
    try:
        return float(v)
    except Exception:
        return 0.0

def _normalize(raw_items: List[Dict[str, Any]]) -> List[NormalizedItem]:
    out: List[NormalizedItem] = []
    for x in raw_items:
        external_id  = str(x.get("id") or x.get("uuid") or x.get("key") or x.get("slug") or "")
        title        = x.get("title") or x.get("name") or x.get("headline") or "(no title)"
        # 画像はよくあるキーを順にチェック
        img_raw      = x.get("image_url") or x.get("image") or x.get("poster_path") or x.get("thumbnail") or x.get("cover")
        image_url    = _img(img_raw)
        summary      = x.get("description") or x.get("overview") or x.get("summary") or ""
        release_date = x.get("release_date") or x.get("published") or x.get("published_at") or x.get("date") or ""
        popularity   = _to_float(x.get("popularity") or x.get("score") or x.get("rating") or x.get("vote_average") or 0.0)
        # ジャンルは文字列/配列/オブジェクトいずれも緩く対応
        genre_val    = x.get("genre") or x.get("genres") or x.get("category") or x.get("categories")
        if isinstance(genre_val, list):
            genre = ", ".join([g.get("name") if isinstance(g, dict) else str(g) for g in genre_val if g])
        elif isinstance(genre_val, dict):
            genre = genre_val.get("name") or genre_val.get("title") or None
        else:
            genre = genre_val or None

        out.append({
            "external_id": external_id,
            "title": title,
            "genre": genre,
            "image_url": image_url,
            "summary": summary,
            "release_date": release_date,
            "popularity": popularity,
        })
    return out

class Provider(ProviderBase):
    def fetch_popular(self, page: int = 1) -> List[NormalizedItem]:
        # エンドポイント/クエリ名は .env から（未設定時は妥当なデフォルト）
        path       = current_app.config.get("PROVIDER_POPULAR_PATH") or "/<your-popular-endpoint>"
        page_key   = current_app.config.get("PROVIDER_PAGE_PARAM") or "page"
        limit_key  = current_app.config.get("PROVIDER_LIMIT_PARAM") or None
        limit_val  = current_app.config.get("PROVIDER_DEFAULT_LIMIT")

        params: Dict[str, Any] = {page_key: page}
        if limit_key and limit_val:
            params[limit_key] = int(limit_val)

        data = _req(path, **params)
        raw_items = _pick_list(data)
        return _normalize(raw_items)

ファイルの説明

あなたのAPIごとの差分（通信・認証・正規化）を 1ファイルに集約。

コードの解説

• _build_headers()： ヘッダー認証が必要なら自動付与
• _maybe_add_query_auth()： クエリ認証が必要なら自動付与
• _req()： ベースURLに対してGET
• normalize： 外部レスポンス → 内部共通形式へ変換（最重要）
• Provider.fetch_popular： 人気/注目/新着エンドポイントにアクセスして配列を正規化（最重要）

自分用に変更する箇所

• なし

10. 取り込みスクリプト（scripts/ingest_catalog.py）

scripts/ingest_catalog.py

import os, sys
sys.path.append(os.path.dirname(os.path.dirname(__file__)))

from app import create_app, db
from app.models.catalog import Catalog
from app.services.providers import get_provider

def upsert(items, provider_name: str):
    try:
        for it in items:
            row = Catalog.query.filter_by(external_id=it["external_id"]).first()
            if not row:
                row = Catalog(external_id=it["external_id"], provider=provider_name)
                db.session.add(row)
            # UPSERT：既存は更新、無ければ作成
            row.title = it.get("title") or "(no title)"
            row.genre = it.get("genre")
            row.image_url = it.get("image_url")
            row.summary = it.get("summary")
            row.release_date = it.get("release_date")
            # popularity が None/文字列でも耐える
            try:
                row.popularity = float(it.get("popularity") or 0.0)
            except (TypeError, ValueError):
                row.popularity = 0.0
        db.session.commit()
    except Exception as e:
        db.session.rollback()
        raise

if __name__ == "__main__":
    app = create_app()
    with app.app_context():
        provider_name = app.config.get("PROVIDER", "")
        pages = int(app.config.get("INGEST_PAGES", 2))
        if not provider_name:
            print("ERROR: .env の PROVIDER が未設定です。")
            sys.exit(1)
        try:
            prov = get_provider()
            all_items = []
            for p in range(1, pages + 1):
                batch = prov.fetch_popular(page=p)
                if not batch:
                    break
                all_items.extend(batch)
            upsert(all_items, provider_name)
            print(f"ingested: {len(all_items)} items from provider={provider_name}")
            sys.exit(0)
        except Exception as e:
            print(f"ERROR: ingest failed: {e}")
            sys.exit(2)

ファイルの説明

API → 正規化配列 → DB に UPSERT する一括取り込み。

コードの解説

• get_provider()： .env の PROVIDER を動的ロード
• external_id で存在チェック → 追加/更新 → commit()
• INGEST_PAGES で取得量を調整

自分用に変更する箇所

• なし

11. レコメンド（app/services/recommend.py）

app/services/recommend.py

# 人気順で上位K件を返す最小レコメンダ。DBのCatalogを読むだけ。
from typing import List, Dict
from flask import current_app
from app.models.catalog import Catalog

class Recommender:
    @property
    def DEFAULT_TOP_K(self) -> int:
        # .env を反映（無ければ9）
        return int(current_app.config.get("DEFAULT_TOP_K", 9))

    def predict(self, top_k: int | None = None) -> List[Dict]:
        k = top_k or self.DEFAULT_TOP_K
        # popularity が NULL の行は0扱いで後ろへ（SQLiteはNULLが最後になる）
        rows = (Catalog.query
                .order_by(Catalog.popularity.desc(), Catalog.created_at.desc())
                .limit(k).all())
        return [self._to_dict(r) for r in rows]

    def _to_dict(self, r: Catalog) -> Dict:
        return {
            "id": r.id,
            "title": r.title,
            "genre": r.genre,
            "image_url": r.image_url,
            "summary": r.summary,
            "release_date": r.release_date,
        }

ファイルの説明

人気順上位K件を返す最小レコメンダ。

コードの解説

• popularity desc, created_at desc で安定ソート
• テンプレ用 dict に整形して返す

自分用に変更する箇所

• ランキング指標を変えたい時のみ

12. 画面ルート（app/routes/web.py）

app/routes/web.py

from flask import Blueprint, render_template, request, current_app
from app.services.recommend import Recommender

web_bp = Blueprint("web", __name__)

@web_bp.get("/")
def top():
    # クエリは文字列なので安全に int 化。別名 k でも受けられるようにしておく。
    top_k = request.args.get("top_k", type=int)
    if top_k is None:
        top_k = request.args.get("k", type=int)

    if top_k is None:
        # .env → config から取得（なければ 9）
        top_k = int(current_app.config.get("DEFAULT_TOP_K", 9))

    items = Recommender().predict(top_k=top_k)
    return render_template("top.html", items=items, top_k=top_k)

ファイルの説明

トップページ / でレコメンド結果をテンプレへ渡す。

コードの解説

• @web_bp.get("/")： ルート登録（GET）
• request.args.get("top_k")： 件数クエリ
• render_template("top.html", ...)： 描画

自分用に変更する箇所

• 文言・既定件数の微調整【任意】

13. テンプレート（app/templates/base.html, app/templates/top.html）

app/templates/base.html

<!doctype html>
<html lang="ja">
  <head>
    <meta charset="utf-8" />
    <title>{% block title %}Demo{% endblock %}</title>
    <meta name="viewport" content="width=device-width, initial-scale=1" />
    <link rel="stylesheet" href="{{ url_for('static', filename='css/style.css') }}">
  </head>
  <body>
    <main class="container">
      {% block content %}{% endblock %}
    </main>
  </body>
</html>

app/templates/top.html

{% extends "base.html" %}
{% block title %}おすすめ（popular）{% endblock %}
{% block content %}
  <h1>おすすめ（popular）</h1>

  <form method="get" class="toolbar">
    <label>件数 <input type="number" name="top_k" min="1" max="50" value="{{ top_k }}"></label>
    <button type="submit">更新</button>
  </form>

  {% if items %}
    <div class="grid">
      {% for it in items %}
        <article class="card">
            {% if it.image_url %}
                <img class="thumb" src="{{ it.image_url }}" alt="{{ it.title }}の画像">
            {% else %}
                <div class="thumb placeholder" aria-label="{{ it.title }}（画像なし）"></div>
            {% endif %}
          <h3>{{ it.title }}</h3>
          <p class="meta">{{ it.genre or "-" }}{% if it.release_date %} / {{ it.release_date }}{% endif %}</p>
          <p class="desc">{{ (it.summary or "")[:100] }}{% if it.summary and it.summary|length > 100 %}…{% endif %}</p>
        </article>
      {% endfor %}
    </div>
  {% else %}
    <p class="empty">データがありません。取り込みを実行してください。</p>
  {% endif %}
{% endblock %}

ファイルの説明

base.html は共通レイアウト。top.html はカードグリッド。

コードの解説

• Jinja2 の extends / block でレイアウト継承
• 画像が無い場合は <img> を省略

自分用に変更する箇所

• UI文言・文字数・配置など調整

14. CSS（app/static/css/style.css）

app/static/css/style.css

.container { max-width: 1040px; margin: 0 auto; padding: 16px; }
.toolbar { margin: 12px 0 20px; display: flex; gap: 8px; align-items: center; }
.grid { display: grid; grid-template-columns: repeat(auto-fill, minmax(180px, 1fr)); gap: 16px; }
.card { border: 1px solid #eee; border-radius: 12px; padding: 12px; background: #fff; }
.card .thumb { width: 100%; height: 270px; object-fit: cover; border-radius: 8px; }
.card h3 { font-size: 1rem; margin: 8px 0 4px; line-height: 1.3; }
.card .meta { color: #666; font-size: .85rem; margin-bottom: 6px; }
.card .desc { color: #333; font-size: .9rem; }
.empty { color: #888; }
.card .thumb.placeholder { width: 100%; height: 270px; border-radius: 8px; background: #f3f3f3; display: block; }

ファイルの説明

• カードUIの最小スタイル。

コードの解説

• CSS Grid で自動段組み
• 画像は object-fit: cover で見栄えキープ

自分用に変更する箇所

• デザイン調整【任意】

15. マイグレーション → 取り込み → 起動

1. マイグレーション（初回のみ）

Windows PowerShell

コマンドプロンプト

$env:FLASK_APP="run.py"
flask --app run.py db init
flask --app run.py db migrate -m "create catalog"
flask --app run.py db upgrade

Mac / Linux

コマンドプロンプト

export FLASK_APP=run.py
flask --app run.py db init
flask --app run.py db migrate -m "create catalog"
flask --app run.py db upgrade

2. 取り込み

コマンドプロンプト

python scripts/ingest_catalog.py
# → ingested: N items from provider=<your_provider>

3. 起動

コマンドプロンプト

python run.py

No changes detected の場合：app/__init__.py に from . import models があるか再確認。

16. 完成ディレクトリ

my_app/
├─ app/
│  ├─ __init__.py
│  ├─ config.py
│  ├─ models/
│  │  ├─ __init__.py
│  │  └─ catalog.py
│  ├─ routes/
│  │  ├─ web.py
│  │  └─ api.py
│  ├─ services/
│  │  ├─ recommend.py
│  │  └─ providers/
│  │     ├─ __init__.py            # ← 動的ロード
│  │     ├─ base.py
│  │     └─ <your_provider>.py     # ← あなたが作る実装
│  ├─ templates/
│  │  ├─ base.html
│  │  └─ top.html
│  └─ static/
│     └─ css/style.css
├─ scripts/
│  └─ ingest_catalog.py
├─ migrations/
├─ .env
├─ run.py
└─ requirements.txt

17. 「自分用に変更する点」最終まとめ

• .env【必須】
  • PROVIDER=<your_provider>
  • PROVIDER_BASE_URL=<YOUR_BASE_URL>
  • PROVIDER_API_KEY=<YOUR_API_KEY>（不要なら空）
• app/services/providers/<your_provider>.py【必須】
  • _req()：エンドポイントと認証方式（クエリ or ヘッダー）
  • _normalize()：外部キー名 → 共通キーへのマッピング
    • 必須：external_id, title, image_url, summary, release_date, popularity（genre 任意）
  • fetch_popular()：人気/注目/新着に相当する一覧を返す（配列抽出キーも設定）
• app/services/providers/__init__.py【作業不要】
  • 動的ロードにより .env の PROVIDER と同名のモジュール <your_provider>.py を自動で読み込み、Provider クラスを実体化します。
• （必要時のみ）モデル変更【慎重】
  • Catalog に列追加 → flask db migrate / upgrade
• 画面 / レコメンド / 取り込み【原則ノータッチ】
  • 文言やUIの微調整は自由