streamlitとnotionを使った在庫管理システムの開発

初めに

所属するラボの在庫管理が微妙だったので、改善のために新規システムを作った。
大まかな概要としては、streamlitで入力した情報をベースに、notionDBに書き加える形にしている。

streamlitの導入

streamlit自体は、普通のライブラリなので、installすればOK(ついでにnotion-clientも導入しておく)

install

pip install streamlit notion-client

また、完成したプログラムファイル自体の起動も通常通り

run streamlit

conda activate yourenv # streamlitを入れた環境を活性化
streamlit run your-app.py

参考リンク

notionAPIの起動

流れとしては、

• インテグレーションの起動
• Notionデータベースの作成
• データベースへのインテグレーション接続
  となる。

参考リンク

1st-インテグレーションの起動

まず、以下のリンクからインテグレーション?を作成する。

リンクを踏むと、インテグレーションを作成、というリンクが出ているはずなのでそこをクリック

※前述の参考リンクから引用
ここで、

• インテグレーション種類
  • アクセス権限の指定？(internal推奨)
• 関連ワークスペース
  • インテグレーションでアクセスするワークスペースを選択
• 名前
  • インテグレーションの名前を入力(見やすい方がよさげ)
• ロゴ
  • ロゴ

を入力することで作成できる。
作成後、内部インテグレーションシークレットが出力されるが、これがnotionAPIで使うAPIトークンとなる。(公開しないよう注意！)
アクセス制限については、APIトークンの下の方にあるのでそこで選択

2nd-Notionデータベースの作成

notionに入り、ページのどこかしらで"/database"と入力して、新規データベースを作成する(インラインフルラインどちらでもOK)

3rd-データベースへのインテグレーション接続

その後、データベースのあるページ(orその親ページ)の右上の三点マークをクリックし、接続(connect)から、先ほど作成したインテグレーションを選択する。
すると、親ページにAPI経由でアクセスできるようになる。

また、データベースの右上にある三点マークから、ビューのリンクをコピーしておく

コピーされたリンクは、

https://www.notion.so/AAAA?v=BBBB&pvs=1

のようになっているが、このAAAの部分があなたのapiトークンでアクセスするためのデータベース自体のIDとなる。
このIDベースでアクセスするのでメモっておくべき

以上で、streamlitの導入と、notionAPIの導入に成功した。
※できなかった場合は公式のドキュメントとかを見てほしい。

streamlitを用いたアプリ作成

ここからのコードはすべてclaude3.7 sonnet thoughtやmercury coderを使って作成しました。

開発前の目標

開発前に、どのようなものにするかを設定しました。
構想としては、製品コードを入力して商品情報を確認、その後入庫数などを入力して在庫変動が起こる、というものにしました。
なので必要となったものは

1. 製品コード入力
2. 製品コードに対応した情報をマスターDBから取得
3. 商品画像による確認
4. 入庫数などの入力
5. 在庫管理DB、ログDBへの記入
6. 商品情報を管理するマスターDB
7. 在庫数を管理する在庫管理DB
8. 在庫変動ログを保管するログDB
9. マスターDBに情報を追加する仕組み
10. 直近ログの表示

の以上になりました。個別に行きます

1 製品コード入力

数値入力用関数があるので、それを使いました。

product_number

product_id = st.number_input("製品番号(product number)を入力してください", 
                                     min_value=0, 
                                     step=1, 
                                     #format="%6d",
                                     )

# 製品コードの保存
st.session_state.product_id = product_id

このコードによって、以下のような入力欄が出ます。

steps=1にすることで、入力される数値を整数(int型)として取得されるようにしています。
これを指定しないと、変数がfloat型として認識されますので注意してください。

2 製品コードに対応した情報をマスターDBから取得

pythonのrequestsライブラリを使いました。
json形式で返ってきます。

product info

# secretから環境変数にNotion APIトークンを設定
os.environ['API_KEY'] = st.secrets["NOTION_API_KEY"]
os.environ['MASTER_DATABASE_ID'] = st.secrets["MASTER_DATABASE_ID"]

# 環境変数からNotion APIトークンを取得
notion_token: str = os.environ['API_KEY']
master_database_id: str = os.environ['MASTER_DATABASE_ID']

# Notion APIクライアントの設定
headers: Dict[str, str] = {
    "Authorization": f"Bearer {notion_token}",
    "Content-Type": "application/json",
    "Notion-Version": "2022-06-28"
}

# NotionデータベースのURL
master_url: str = f"https://api.notion.com/v1/databases/{master_database_id}/query"

# product_idをベースにデータベースを探索する際のコード
query: Dict[str, Any] = {
        "filter": {
            "property": "product number",# notion DBの項目と対応
            "number": { # notion DBの項目のプロパティと対応
                "equals": product_id # 一致で検索
            }
        }
    }
response = requests.post(master_url, # アクセス先のアドレス
                        headers=headers, # 共通のヘッダー
                        data=json.dumps(query), # クエリ
                        verify=True,# HTTPS通信するための明示
                        )
data = response.json() # json変換
results = data.get("results", [])

この時、queryのpropertyは、notionDB上のプロパティと対応している必要がある。
今回は、数値を扱うようにしているので、プロパティの部分はnumberとなっている。

この、DB上のプロパティとqueryの部分で一致していないとエラーを吐くので注意。(後で解説予定)

3 商品画像による確認

requestから受けた結果から、画像情報を取り出して取得しました。
データ構造としては以下のようになっていました。

[
    0:{
        "properties":{
            "image":{
            "id":"image_id"
            "type":"files"
                "files":[
                        0:{
                            "name":"526699_0.jpg"
                            "type":"file"
                            "file":{
                            "url":"image_url"
                            "expiry_time":"date"
                            }
                        }
                        ]
            }
            }
        }
]

ですので、ここから画像を抽出する際は以下のようになりました。

product_image = results[0]["properties"]["image"]["files"][0]["file"]["url"]
image = Image.open(requests.get(product_image, stream=True).raw)

ここから、画像を表示し、製品画像が一致するかどうかは以下の関数で確かめられるようにしました。
この関数では、製品コード、画像、名前、会社をユーザーに確認してもらい、正しければ"confirm"ボタンを、違っていれば"refuse"ボタンを押すようにしています。

image_popup

def confirm_product(
        product_id: str, 
        product_image: Image.Image, 
        product_name: str, 
        company: str
        ) -> bool:
    """
    製品情報を確認するためのポップアップを表示する関数。

    Args:
        product_id: 製品ID。
        product_image: 製品画像。
        product_name: 製品名。
        company: 会社名。

    Returns:
        確認された場合はTrue、拒否された場合はFalse。
    """
    st.image(
        product_image,# 製品画像
        width=200,# 画像表示の幅
    )

    st.write(f"この製品を確認してください。  \n Product number: {product_id}  \n Product name: {product_name}  \n Company: {company}")
    # 画像のキャプション

    if 'product_confirmed' not in st.session_state:
        st.session_state.product_confirmed = False
    # 変数の初期化
    # 最終的に製品画像が正しければTrueに、間違っていればFalseになる

    col1, col2 = st.columns(2)
    with col1:
        if st.button("Confirm"):
            st.session_state.product_confirmed = True
            return True
    with col2:
        if st.button("Refuse"):
            st.session_state.product_confirmed = False
            return False

    return st.session_state.product_confirmed
    
confirmed = confirm_product(product_id, image, product_name, company)

実際のアプリ上の画面は以下の通りです。

4 入庫数などの入力

製品コードと同じように、st.number_inputを使いました

update_inventory

st.write(f"製品番号 {st.session_state.product_id} ({st.session_state.product_name}) の在庫状況を設定します。")

action = st.selectbox("Select Action", ["入庫", "開封", "廃棄"])
quantity = st.number_input("個数を入力してください", min_value=1, value=1)
user_name = st.text_input("ユーザー名を入力してください", "bneiea")
storages = st.multiselect("保管場所を選択してください", ["572", "573", "575", "576", "587", "588"], [st.session_state.storage])
note = st.text_area("備考を入力してください", "")

if st.button("Update Inventory"):
    # 製品コードと名称は、st.sessionに設定した値を採用
    product_id = st.session_state.product_id
    product_name = st.session_state.product_name

    if update_inventory(product_id, action, quantity):
        st.success(f"Inventory updated for {product_id}")
        # ログへの追加
        log_action(
                zaiko_database_id=log_database_id,
                product_number=product_id,
                product_name=product_name,
                status=action,
                quantity=quantity,
                user_name=user_name,
                note=note,
                storages=storages,
            )

このコードによって、以下のような画面が表示され、入力後にUpdate Inventoryボタンを押すことで、在庫データベースが更新されます。

5 在庫管理DB、ログDBへの記入

requestライブラリを使いました。

update_inventory

def update_inventory(
    product_id: str, 
    action: str, 
    quantity: int = 1, 
    url: str = zaiko_url
) -> bool:
    """
    Notionデータベースの在庫情報を更新する関数。

    Args:
        product_id: 製品ID。
        action: 実行されたアクション（"入庫", "開封", "廃棄"）。
        quantity: 更新する数量。
        url: 在庫データベースのURL。

    Returns:
        更新が成功した場合はTrue、失敗した場合はFalse。
    """
    # 製品番号をもとにして、在庫の初期値を取得
    query: Dict[str, Any] = {
        "filter": {
            "property": "product number",
            "number": {
                "equals": product_id
            }
        }
    }
    response = requests.post(url, headers=headers, data=json.dumps(query), verify=True)
    data = response.json()
    results = data.get("results", [])
    if results:
        # 初期値の取得
        page_id = results[0]["id"]
        inventory = int(results[0]["properties"]["stock quantity"]["number"])
        if action == "入庫":
            inventory += quantity
        elif action == "開封":
            inventory -= quantity
        elif action == "廃棄":
            inventory -= 0  # 箱単位で管理するなら、この項目による増減は不要になるはず。
        update_query: Dict[str, Any] = {
            "properties": {
                "stock quantity": {
                    "number": inventory
                }
            }
        }
        update_url = f"https://api.notion.com/v1/pages/{page_id}"
        requests.patch(update_url, headers=headers, data=json.dumps(update_query), verify=True)
        #データベースを更新する場合は、patch関数を使う。
        return True
    return False

また、この関数でlogデータベースへの追加をしました。

log_action

def log_action(
    zaiko_database_id: str,
    product_number: int,
    product_name: str,
    status: str,
    quantity: int,
    user_name: str,
    note: str = "",
    storages: List[str] = ["572", "573", "575", "576", "587", "588"],
) -> bool:
    """
    Notionデータベースにログ情報を追加する関数。

    Args:
        zaiko_database_id: ログを記録するデータベースID。
        product_number: 製品番号。
        product_name: 製品名。
        status: 実行されたアクション（"入庫", "開封", "廃棄"）。
        quantity: 数量。
        user_name: ユーザー名。
        note: 備考。
        storages: 保管場所のリスト。

    Returns:
        ログの追加が成功した場合はTrue、失敗した場合はFalse。
    """
    dt_now = datetime.datetime.now()

    log_entry: Dict[str, Any] = {
        "parent": {"database_id": zaiko_database_id},
        "properties": {
            "date": {"date": {"start": dt_now.strftime("%Y-%m-%d")}},
            "product name": {"title": [{"text": {"content": product_name}}]},
            "product number": {"number": product_number},
            "quantity": {"number": quantity},
            "status": {"select": {"name": status}},
            "user_name": {"rich_text": [{"text": {"content": user_name}}]},
            "storage": {"multi_select": [{"name": storage} for storage in storages]},
            "note": {"rich_text": [{"text": {"content": note}}]},
        }
    }

    response = requests.post(
        "https://api.notion.com/v1/pages", 
        headers=headers, 
        data=json.dumps(log_entry), 
        verify=True,
    )
    if response.status_code == 200:
        return True
    else:
        print(response.text)
        return False

6 商品情報を管理するマスターDB

DBの構造についてです。以下の項目を持つようにしました。

項目	プロパティ	内容
product_number	number	製品番号
product_name	rich_text	製品の名称
company	rich_text	会社名
product info	rich_text	製品説明
where	multi_select	保管場所
image	files	商品画像
note	rich_text	備考

7 在庫数を管理する在庫管理DB

構造は以下の通りです。

項目	プロパティ	内容
product_number	number	製品番号
product_name	rich_text	製品の名称
stok quantity	number	在庫数
where	multi_select	保管場所
image	files	商品画像
note	rich_text	備考

8 在庫変動ログを保管するログDB

在庫追加ログ記録DBです。

項目	プロパティ	内容
date	date	登録年月日
product_number	number	製品番号
product_name	rich_text	製品の名称
quantity	number	追加在庫数
status	select	在庫設定
user_name	rich_text	在庫追加者
storage	multi_select	保管場所
note	rich_text	備考

9 マスターDBに情報を追加する仕組み

製品番号が記録されていない場合は、新規に登録する仕組みも作りました。
コードは以下の通りです。長いのでたたみます。
画面としてはこんな感じです。

マスターへの追加コード

add_master

elif st.session_state.step == 'add_master':# マスターへの登録モードへ移行
    st.write("マスター登録フォーム")
    product_id = st.session_state.product_id
    st.write(f"製品番号: {product_id}")
    product_name = st.text_input("製品名を入力してください")
    company = st.text_input("会社名を入力してください")
    product_info = st.text_input("製品情報を入力してください")
    storages = st.multiselect("保管場所を選択してください", ["572", "573", "575", "576", "587", "588"], ["573"])
    note = st.text_input("備考を入力してください")
    image_url = st.text_input("可能なら画像URLを入力してください")
    #st.session_state.image = False
    if image_url:
        image = Image.open(requests.get(image_url, stream=True).raw)
        st.image(image, caption="Image URL", use_container_width=True)

    if st.button("マスター登録"):
        master_entry: Dict[str, Any] = {
            "parent": {"database_id": master_database_id},
            "properties": {
                "product number": {"number": product_id},
                "product name": {"rich_text": [{"text": {"content": product_name}}]},
                "company": {"rich_text": [{"text": {"content": company}}]},
                "product info":{"title": [{"text": {"content": product_info}}]},
                "where": {"multi_select": [{"name": storage} for storage in storages]},
                "note" : {"rich_text": [{"text": {"content": note}}]},
            }
        }
        if image_url:
            master_entry["properties"]["image"] = {"files": [{"name": product_name, "external": {"url": image_url}}]}

        response = requests.post(
                "https://api.notion.com/v1/pages", 
                headers=headers, 
                data=json.dumps(master_entry),
                verify=True,
        )
        if response.status_code == 200:
            st.success(f"マスター登録が完了しました。")
            # 5秒待ってから自動で最初の画面へ移動
            wait_time = 5
            time.sleep(wait_time)
            st.session_state.step = 'scan'
            st.rerun()
        else:
            st.write(response.text)
            # 5秒待ってから自動で最初の画面へ移動
            st.error(f"マスター登録に失敗しました。再試行してください。")
            st.session_state.step = 'add_master'
            wait_time = 10
            time.sleep(wait_time)
            st.rerun()

notionAPIでは、直接画像を扱うということが出来ないようです。(20250327確認)
https://developers.notion.com/docs/working-with-files-and-media

Uploading files and media via the Notion API
The API currently does not support uploading new files.

基本的に、画像はリンクで管理されているので、画像へのリンクをコピーなどして得られたリンクを貼るのがいいかと思います。

10 直近ログの表示

今回は、以下のように、直近5件の在庫追加ログを表示するようにしました。

なぜかjsonへの直接変換がうまくいかなかったこともあり、以下のような汚いコードが出力されました。

ログ表示関数

流れとしては、データベースから得られたデータをjsonに直し、そこから地道にdataframeにしています。

recent log

def process_notion_data(database_id: str, n: int, headers: Dict[str, str]) -> pd.DataFrame:
    """Processes a list of Notion API responses into a pandas DataFrame.

    Args:
        database_id: ID of log DB

    Returns:
        A pandas DataFrame with the processed data. Returns an empty DataFrame if input is invalid or empty.
    """
    url = f"https://api.notion.com/v1/databases/{database_id}/query"
    response = requests.post(url, headers=headers, data=json.dumps({"page_size": n}), verify=True)
    notion_data = response.json().get("results")
    # ログデータの取得
    if not notion_data:
        return pd.DataFrame()
    
    processed_data = []
    for row in notion_data:
        properties = row["properties"]
        processed_row = {}
        for key, value in properties.items():
            # Handle different property types
            if value.get("type") == "date":
                processed_row[key] = value["date"].get("start")
            elif value.get("type") == "title":
                processed_row[key] = value["title"][0]["text"]["content"] if value["title"] else ""
            elif value.get("type") == "number":
                processed_row[key] = value["number"]
            elif value.get("type") == "select":
                processed_row[key] = value["select"]["name"]
            elif value.get("type") == "rich_text":
                processed_row[key] = value["rich_text"][0]["text"]["content"] if value["rich_text"] else ""
            elif value.get("type") == "multi_select":
                processed_row[key] = ", ".join([item["name"] for item in value["multi_select"]])
            else:
                processed_row[key] = str(value) # Handle other types as strings

        processed_data.append(processed_row)

    return pd.DataFrame.from_records(processed_data)

# Example usage: Get the top 5 rows from your database
top_5_rows = process_notion_data(database_id=log_database_id, n=5, headers=headers)
if top_5_rows is not None:
    st.dataframe(
        top_5_rows[["date", "product number", "product name", "user_name", "storage", "status", "quantity", "note"]],
        hide_index=True,
        #use_container_width=True,
    )
else:
    st.write("Failed to retrieve data from Notion.")

コードまとめ

コードとしては以上になります。
コードはここにあげました。

補足色々

ここからは、補足です。

バーコード読み取りとその方法

当初は、製品のバーコードを読み、製品コードに変換して、在庫数変動を記録する形にしていました。
バーコード取得には、pyzbarというライブラリが必要となり、以下のようにバーコードを取得することが出来ます。

camera input

# pip install pyzbar
from pyzbar import pyzbar
import streamlit as st

def decode_barcode(img: Image.Image):
    """バーコードを読み取り、デコード結果を返す"""
    if not isinstance(img, Image.Image):
        raise ValueError("img must be a PIL.Image.Image")
    # バーコードを読み取り
    decoded_objects = pyzbar.decode(img)
    if not decoded_objects:
        return None
    if len(decoded_objects) > 1:
        st.warning("複数のバーコードが検出されました。")
    # デコード 
    return [[obj.data, obj.type] for obj in decoded_objects]

img = st.camera_input("バーコードを読み取る")
if img is not None:
    # バーコードを読み取り
    barcodes = decode_barcode(img)
    barcodes_length = len(barcodes) if barcodes else 0
    #　バーコードが複数読み込まれた場合の対応
    if barcodes_length > 1:
        barcode = st.selectbox("選択してください:", [b[0].decode("utf-8") for b in barcodes])
        st.write("選択されたバーコード:", barcode)
        # バーコードを製品コードに変換
        product_number = convert_from_barcode_to_product_number(barcode)

    elif barcodes_length == 1:
        barcode = barcodes[0][0].decode("utf-8")
        st.write("バーコード読み取り結果:", barcode)
        product_number = convert_from_barcode_to_product_number(barcode)
else:# バーコード読み取りがうまくいかなかった場合に直接製品コードを入力する
    st.warning("バーコードが検出できませんでした。直接製品番号を入力してください")
    product_number = st.number_input("製品番号(product number)を入力してください", min_value=0, step=1, format="%6d")
                

以上のコードによって、バーコードを取得できますが、streamlit cloud上ではうまくいきませんでした。
pyzbar自体は、ZBarなるライブラリをベースに動いているらしいのですが、その導入をstreamlit cloud上でやる方法が分からず、あきらめました。

streamlit cloudの使い方

アプリのデプロイについては、以下のリンクの通りです。
(Advanced settingsに、secretsを設定するところがあります)

APIキーを安全に使う方法

自分のpcでローカル実行場合は、例えばos.environで取得すればいいんですが、streamlit cloudを使う場合、コードは基本的にオープンになってる前提で考える必要があり、雑にコード内部に記載すると言ったことはできません。
ですので、別の機能を利用する必要があります。
streamlit cloudを使う場合、アプリのデプロイタイミングで、secretとして、キーと変数を入力することが出来ますので、これによって安全に保APIキーを使うことができます。
設定方法としては、deployの時に表示されるadvanced settingsを開き、secretsに

API_KEY = 'your-api-key'
ENV-KEY = 'ENV-KEY'

のように追加し、コード上で以下のようにすることで取得できます。

os.environ['API_KEY'] = st.secrets['API_KEY']

これにより、apiキーを安全に取り扱うことができます。

requestsライブラリによる通信の安全性

よくわかりませんでしたが、この記事を読む限り、verify=Falseにしない限りはHTTPS通信をしていると思ってよさそうです。

また、ドキュメントには以下のように書いてありました。

verify – (optional) Either a boolean, in which case it controls whether we verify the server’s TLS certificate, or a string, in which case it must be a path to a CA bundle to use. Defaults to True. When set to False, requests will accept any TLS certificate presented by the server, and will ignore hostname mismatches and/or expired certificates, which will make your application vulnerable to man-in-the-middle (MitM) attacks. Setting verify to False may be useful during local development or testing.

verify
SSL Verification default. Defaults to True, requiring requests to verify the TLS certificate at the remote end. If verify is set to False, requests will accept any TLS certificate presented by the server, and will ignore hostname mismatches and/or expired certificates, which will make your application vulnerable to man-in-the-middle (MitM) attacks. Only set this to False for testing.

となっていました。なので、おそらく安全性という意味では問題ないと思います。(多分！TLS証明が分からないけど！)

まとめ

今回は以上になります。
作ってみて、webアプリケーションにおける画面表示(フロントエンド)の勉強をしたくなりました。
また、安全な通信という概念に気を付けようと思いました。
ちなみに、このアプリは最終的に採用されなくなりました。

個人的な布教

github copilotにも関係ある言語モデルについての以下の記事が非常に面白かったです。

読んだ限り、next token predictionであることが制約になっているように感じたので、今回使った拡散言語モデル(diffusion language model)に夢がありそうに感じました。
コードはここで試せます。