みなさん、テレビ見てますか
どっかのテレビ局は、不祥事で大変みたいですが、まだまだ、電波放送の役目は終わってませんね。
ということで、番組改編の時期に新番組の予約とかを、せっせっとやりましょ
我らのEPGステーションでね
でも、ルール作るのって面倒ですよね。
なので、良い感じで新番組との予約ルールを作ってくれるのをサポートしてくれるWebツールを作成しました。
まずは、番組表をキーワードで検索して、[新]とかってやると新番組がひっかかりますよね。
適当な番組をチェックして、予約フォームを表示すると、AIが連続番組用にキーワードと保存ディレクトリを作成してくれちゃいます。
後は、エンコードモードを選択して予約を作成すれば終了。
結構、キーワードを考えるのが面倒だったんですよね。
でも、OCIのAI を使って、良い感じのキーワードが生成されるので、楽ちんですよ。
ついでに、保存ディレクトリも
こうやって、ちっちゃいとこにも AI が使えたりして便利になりました。
ということで、ソースです。
import streamlit as st
import urllib.parse
import urllib.request
import json
import datetime
import math
import pytz
import pandas as pd
import time
import oci
from oci.config import from_file
from oci.generative_ai_inference import GenerativeAiInferenceClient
from oci.generative_ai_inference.models import (
ChatDetails,
CohereChatRequest,
OnDemandServingMode
)
config = oci.config.from_file("~/.oci/config", "DEFAULT")
COMPARTMENT_ID = "ocid1.compartment.oc1..aaaaaaaamgmw22hogecwqnunirb3urhoger4ihdgoilkdjkv2sabokaq5svc"
# Generative AIクライアントを作成
aiclient = GenerativeAiInferenceClient(config=config)
MODEL_DEF = "cohere.command-a-03-2025"
# EPG Station のAPIエンドポイント
EPG_STATION_API_BASE = "http://localhost:8888/api/schedules"
EPG_STATION_RESERVES_API = "http://localhost:8888/api/reserves"
# タイムゾーン設定
jst = pytz.timezone('Asia/Tokyo')
st.set_page_config(page_title="EPG Station 番組表ビューア", layout="wide")
st.title("EPG Station 番組表ビューア")
# --- セッション状態の初期化 ---
if 'program_list' not in st.session_state:
st.session_state.program_list = []
if 'current_page_schedule' not in st.session_state:
st.session_state.current_page_schedule = 1
if 'page_size_schedule' not in st.session_state:
st.session_state.page_size_schedule = 25
if 'selected_program' not in st.session_state:
st.session_state.selected_program = None
if 'show_reservation_form' not in st.session_state:
st.session_state.show_reservation_form = False
# ジャンル定義
GENRE_MAP = {
0:'ニュース/報道', 1:'スポーツ', 2:'情報/ワイドショー', 3:'ドラマ', 4:'音楽',
5:'バラエティ', 6:'映画', 7:'アニメ/特撮', 8:'ドキュメンタリー/教養',
9:'劇場/公演', 10:'趣味/教育', 11:'福祉', 12:'スポーツ(CS)', 13:'映画(CS)',
14:'拡張', 15:'その他'
}
# サイドバーに検索条件を配置
st.sidebar.header("番組検索条件")
# 検索キーワード
search_keyword = st.sidebar.text_input("キーワード", value="[新]", key="schedule_keyword_input")
# 検索期間
day_range = st.sidebar.slider("検索期間 (日)", 1, 7, 7, key="schedule_day_range")
# 放送種別
st.sidebar.subheader("放送種別")
col_gr, col_bs, col_cs, col_sky = st.sidebar.columns(4)
with col_gr:
gr_enabled = st.checkbox("GR", value=True, key="schedule_gr_checkbox")
with col_bs:
bs_enabled = st.checkbox("BS", value=False, key="schedule_bs_checkbox")
with col_cs:
cs_enabled = st.checkbox("CS", value=False, key="schedule_cs_checkbox")
with col_sky:
sky_enabled = st.checkbox("SKY", value=False, key="schedule_sky_checkbox")
# 無料放送
is_free = st.sidebar.checkbox("無料放送のみ", value=False, key="schedule_is_free_checkbox")
# ジャンル選択
selected_genre_name = st.sidebar.selectbox(
"ジャンル",
options=["指定なし"] + list(GENRE_MAP.values()),
index=0,
key="schedule_genre_select"
)
selected_genre_id = None
if selected_genre_name != "指定なし":
genre_id_map = {v: k for k, v in GENRE_MAP.items()}
selected_genre_id = genre_id_map.get(selected_genre_name)
# 検索ボタン
search_programs_button = st.sidebar.button("番組表を取得", use_container_width=True, key="get_schedule_button")
# --- 番組検索関数 ---
@st.cache_data(ttl="1h")
def fetch_and_filter_programs(
day_range_val, is_free_val, gr_val, bs_val, cs_val, sky_val, genre_id_val, keyword_val
):
t1 = int(datetime.datetime.now(jst).timestamp() * 1000)
t2 = t1 + day_range_val * 3600 * 24 * 1000
params = {
"startAt": t1,
"endAt": t2,
"isHalfWidth": False,
"isFree": is_free_val,
"GR": gr_val,
"BS": bs_val,
"CS": cs_val,
"SKY": sky_val
}
query_string = urllib.parse.urlencode(params)
url_get = f"{EPG_STATION_API_BASE}?{query_string}"
try:
with urllib.request.urlopen(url_get, timeout=10) as response:
json_data = json.load(response)
all_programs = []
for chanObj in json_data:
channel = chanObj["channel"]
for prgObj in chanObj["programs"]:
if genre_id_val is not None and prgObj.get("genre1") != genre_id_val:
continue
if keyword_val and keyword_val not in prgObj.get("name", ""):
continue
ctype = channel.get("channelType")
if ctype == "GR" and not gr_val: continue
if ctype == "BS" and not bs_val: continue
if ctype == "CS" and not cs_val: continue
if ctype == "SKY" and not sky_val: continue
s_jst_str = "N/A"
e_jst_str = "N/A"
if "startAt" in prgObj and "endAt" in prgObj:
try:
s_jst = datetime.datetime.fromtimestamp(prgObj["startAt"] / 1000, tz=datetime.timezone.utc).astimezone(jst)
e_jst = datetime.datetime.fromtimestamp(prgObj["endAt"] / 1000, tz=datetime.timezone.utc).astimezone(jst)
s_jst_str = s_jst.strftime('%Y-%m-%d %H:%M:%S')
e_jst_str = e_jst.strftime('%Y-%m-%d %H:%M:%S')
except Exception as e:
st.warning(f"日時変換エラー (ID: {prgObj.get('id', 'N/A')}): {e}")
all_programs.append({
"日時 (開始)": s_jst_str,
"日時 (終了)": e_jst_str,
"ジャンル": GENRE_MAP.get(prgObj.get("genre1"), "N/A"),
"種別": channel.get("channelType", "N/A"),
"チャンネル": channel.get("name", "N/A"),
"番組名": prgObj.get("name", "N/A"),
"説明": prgObj.get("description", "N/A"),
"programId": prgObj.get("id"),
"channelId": channel.get("id"),
"startAt": prgObj.get("startAt"),
"endAt": prgObj.get("endAt"),
"genre": prgObj.get("genre1"),
"subGenre": prgObj.get("subGenre1"),
"isFree": prgObj.get("isFree"),
})
return all_programs
except urllib.error.URLError as e:
st.error(f"EPG Stationへの接続エラー: {e.reason}。EPG Stationが実行中か、アドレスが正しいか確認してください。")
return []
except json.JSONDecodeError:
st.error("EPG Stationからの応答が不正です。APIのレスポンス形式を確認してください。")
return []
except Exception as e:
st.error(f"番組取得中に予期せぬエラーが発生しました: {e}")
return []
# --- 予約ルールをPOSTする関数 ---
# これは別途定義が必要
def post_reserve_rule(rule_payload):
rule_url = "http://localhost:8888/api/rules" # EPG StationのルールAPIエンドポイント
headers = {"Content-Type": "application/json"}
try:
req = urllib.request.Request(
rule_url,
json.dumps(rule_payload, ensure_ascii=False).encode('utf-8'),
headers,
method="POST"
)
with urllib.request.urlopen(req, timeout=10) as f:
if f.status == 200 or f.status == 201:
return True
else:
st.error(f"ルール登録APIエラー: ステータスコード {f.status}")
return False
except urllib.error.URLError as e:
st.error(f"EPG StationルールAPIへの接続エラー: {e.reason}")
return False
except Exception as e:
st.error(f"ルール登録中に予期せぬエラーが発生しました: {e}")
return False
# 結果テキストをAIで、自然言語へ変換する
def parse_talking( msg ) :
# チャットリクエストの作成
chat_request = CohereChatRequest(
message=msg,
max_tokens=500,
temperature=0.6,
is_echo=True,
is_stream=False
)
# サービングモードの指定（利用モデルIDを正しく指定）
serving_mode = OnDemandServingMode(
model_id=MODEL_DEF
)
# チャット詳細情報をまとめる
chat_details = ChatDetails(
compartment_id=COMPARTMENT_ID,
chat_request=chat_request,
serving_mode=serving_mode
)
# チャットAPIを呼び出す
response = aiclient.chat(chat_details)
# ここに音声からのテキスト変換結果を解析するロジックを追加
return response.data.chat_response.text
# --- UIロジック ---
if search_programs_button:
st.session_state.program_list = []
st.session_state.current_page_schedule = 1
st.session_state.show_reservation_form = False
with st.spinner("番組表を取得中..."):
st.session_state.program_list = fetch_and_filter_programs(
day_range, is_free, gr_enabled, bs_enabled, cs_enabled, sky_enabled, selected_genre_id, search_keyword
)
if st.session_state.program_list:
st.success(f"{len(st.session_state.program_list)} 件の番組が見つかりました。")
else:
st.info("指定された条件で見つかった番組はありませんでした。")
# 番組リストの表示
if st.session_state.program_list:
st.subheader("番組検索結果")
total_records = len(st.session_state.program_list)
total_pages = math.ceil(total_records / st.session_state.page_size_schedule)
# ページングコントロール
col_prev_sch, col_current_page_info_sch, col_next_sch = st.columns([1, 2, 1])
with col_prev_sch:
if st.button("⏪ 前のページ", disabled=(st.session_state.current_page_schedule <= 1), key="prev_page_btn_sch"):
st.session_state.current_page_schedule -= 1
st.rerun()
with col_current_page_info_sch:
st.markdown(f"**ページ {st.session_state.current_page_schedule} / {total_pages}** ({total_records}件中)")
with col_next_sch:
if st.button("⏩ 次のページ", disabled=(st.session_state.current_page_schedule >= total_pages), key="next_page_btn_sch"):
st.session_state.current_page_schedule += 1
st.rerun()
# 表示するレコードの範囲を計算
start_idx = (st.session_state.current_page_schedule - 1) * st.session_state.page_size_schedule
end_idx = min(start_idx + st.session_state.page_size_schedule, total_records)
current_page_programs = st.session_state.program_list[start_idx:end_idx]
# 表示用のデータフレームを作成（内部データは別途保持）
df_display = pd.DataFrame(current_page_programs)
display_columns = ["日時 (開始)", "日時 (終了)", "ジャンル", "種別", "チャンネル", "番組名", "説明"]
df_for_selection = df_display[display_columns].copy()
if not df_for_selection.empty:
# dataframeのon_selectを使用して行選択を可能にする
selection = st.dataframe(
df_for_selection,
hide_index=True,
use_container_width=True,
on_select="rerun",
selection_mode="single-row",
key="program_selector"
)
# 選択された行のインデックスを取得
selected_indices = selection.selection.rows if selection.selection.rows else []
# 予約ボタンと選択された番組の表示
if selected_indices:
st.write(f"**{len(selected_indices)}件の番組が選択されています**")
col1, col2 = st.columns([1, 4])
with col1:
if st.button("🔴 予約フォームを表示", type="primary"):
# 最初に選択された番組を予約対象とする
selected_program_idx = start_idx + selected_indices[0]
st.session_state.selected_program = st.session_state.program_list[selected_program_idx]
st.session_state.show_reservation_form = True
st.rerun()
with col2:
selected_program_names = [current_page_programs[idx]["番組名"] for idx in selected_indices]
st.write("選択された番組: " + ", ".join(selected_program_names[:3]) + ("..." if len(selected_program_names) > 3 else ""))
else:
st.info("表示する番組がありません。")
# 予約フォーム表示
if st.session_state.show_reservation_form and st.session_state.selected_program:
st.divider()
st.subheader("🔴 録画予約")
program = st.session_state.selected_program
# 選択された番組の詳細表示
with st.expander("📺 予約する番組の詳細", expanded=True):
col1, col2 = st.columns(2)
with col1:
st.write(f"**番組名:** {program['番組名']}")
st.write(f"**チャンネル:** {program['チャンネル']}")
st.write(f"**ジャンル:** {program['ジャンル']}")
with col2:
st.write(f"**開始時刻:** {program['日時 (開始)']}")
st.write(f"**終了時刻:** {program['日時 (終了)']}")
st.write(f"**放送種別:** {program['種別']}")
if program['説明'] != "N/A":
st.write(f"**番組説明:** {program['説明']}")
# 予約設定フォーム
st.write("### 録画設定")
# LLM で、番組名からええ感じにキーワード候補を作成
programname = program["番組名"]
genrename = program["ジャンル"]
msg = f"「{programname}」は、テレビ番組名です。この番組名から検索文字列を作成して下さい。番組のジャンルは、{genrename} となります。番組は、連続して放送されることもことがあるので、そういった要素は検索文字列に含めないで下さい。検索文字列は、番組を検索する時の部分一致に使用するので余計な文字列を含めないで下さい。もし、「#1」等の番組回を連想させる数値があった場合、その前後の文字列で一意識別となりそうな文字列を採用して下さい。ジャンル、無駄な記号、[新] [字] といった要素は、含めないで下さい。結果コンテンツだけでを出力し、コメントや説明は不要です。1行の文字列で出力して下さい。"
preword = parse_talking( msg )
msg = f"'{preword}' を、ディレクトリパスに利用出来る文字列に変換して下さい。結果コンテンツだけでを出力>し、コメントや説明は不要です。1行の文字列で出力して下さい。"
predirectory = parse_talking( msg )
col1, col2 = st.columns(2)
with col1:
encode_mode = st.selectbox(
"エンコードモード",
options=["H.264HardLow", "H.264Hard", "H.265Low", "H.265", "無変換"],
index=0,
help="録画時のエンコード方式を選択してください"
)
save_directory = st.text_input(
"保存ディレクトリ",
value=predirectory,
help="空白の場合はデフォルトディレクトリに保存されます",
placeholder="directory"
)
with col2:
keyword = st.text_input(
"キーワード",
value=preword,
help="この文字列で番組を検索します。",
placeholder="keyword"
)
# 予約実行ボタン
st.write("### 予約の実行")
col1, col2, col3 = st.columns([1, 1, 2])
with col1:
if st.button("✅ 予約を作成", type="primary"):
with st.spinner("予約を作成中..."):
ctype = program["種別"]
directory = save_directory
#予約情報
reserveinfo = {
"isTimeSpecification": False,
"searchOption": {
"keyword": keyword,
"keyCS": False,
"keyRegExp": False,
"name": True,
"description": False,
"extended": False,
"ignoreKeyCS": False,
"ignoreKeyRegExp": False,
"ignoreName": False,
"ignoreDescription": False,
"ignoreExtended": False,
"GR": True if ctype == "GR" else False,
"BS": True if ctype == "BS" else False,
"CS": True if ctype == "CS" else False,
"SKY": True if ctype == "SKY" else False,
"genres": [
{
"genre": program["genre"],
"subGenre": program["subGenre"]
}
],
"isFree": program["isFree"],
},
"reserveOption": {
"enable": True,
"allowEndLack": True,
"avoidDuplicate": False
},
"saveOption": {
"directory": directory
},
"encodeOption": {
"mode1": encode_mode,
"directory1": directory,
"isDeleteOriginalAfterEncode": True
}
}
success = post_reserve_rule(reserveinfo)
if success:
st.success("ルールを登録しました。")
st.balloons()
# 予約成功後はフォームを閉じる
st.session_state.show_reservation_form = False
st.session_state.selected_program = None
with st.spinner("..."):
time.sleep(4)
st.rerun()
else:
st.error(message)
with col2:
if st.button("❌ キャンセル"):
st.session_state.show_reservation_form = False
st.session_state.selected_program = None
st.rerun()
st.caption("※ EPG Station が http://localhost:8888 で動作している必要があります。")
最近、はまってる streamlit で作ったので、適当に起動すれば動きます。
streamlit run hogehoge.py --server.port 8501 --server.enableCORS false
AI といっしょに適当に作ったので、バグってるところもあるかと思いますが、大目に見てやって下さい。
もし、OCI AI が使えないなら、ChatGPT とか、Gemeni CLI とかでも良いでしょうね。
頑張って実装してみましょうね。
