update, add api modules

ja.README.md

@@ -0,0 +1,61 @@
+---
+title: book-thumb-gen
+emoji: 📙
+colorFrom: gray
+colorTo: yellow
+sdk: gradio
+sdk_version: 4.32.1
+app_file: src/app/run.py
+pinned: true
+license: mit
+---
+
+# Shoei API WebUI
+
+この改良版には以下が含まれます:
+
+1. **コード構造の改善**: クラスと関数が明確に定義され、メソッドが論理的に分離されています。
+
+2. **例外処理**: `get_thumbnail` メソッドにエラー処理を追加しました。
+
+3. **非同期サポート**: API 呼び出しの非同期処理を改善しました。
+
+4. **テスト**: `pytest` を使用して基本的なテストを追加しました。
+
+5. **動的 HTML 生成**: 本のサムネイルと詳細の動的 HTML 生成を改善しました。
+
+## 開始
+
+```sh
+python src/app/run.py
+```
+
+`pytest` を使用してテストを実行します:
+
+```sh
+pytest test/shoei_api.py
+```
+
+必要な依存関係を必ずインストールしてください:
+
+```sh
+pip install gradio aiohttp pytest
+```
+
+### 国立国会図書館 API
+国立国会図書館サーチで提供されているAPIは主に次の5つです。
+
+- 検索用API(SRU): 書名や著者、ISBNなど様々なメタデータから検索をすることができる,GETパラメータで問い合わせるとXMLで情報を返す。
+- 検索用API(OpenSearch;XML): 
+- 検索用API(OpenURL;HTML): 
+- ハーベスト用API(OAI-PMH): サービス間でのメタデータを交換するためのプロトコル用API
+- 書影API: 国会図書館サーチが所持している書影の画像を取得できます。ISBNをGETパラメータにつけて問い合わせると画像データを返す。
+
+### OpenBD API
+
+### Calil API
+- [calil.jp/api/dashboard/](https://calil.jp/api/dashboard/?register=true)
+
+## 関連
+- https://www.ndl.go.jp/jp/service/rssemag.html
+- https://www.hanmoto.com/permission-for-use/free-to-use

requirements.txt

@@ -1,4 +1,7 @@
 aiohttp==3.8.4
+beautifulsoup4==4.12.3
 gradio==4.36.1
 isbnlib==3.10.14
-pytest==7.3.1
+pandas==1.5.3
+python-dotenv==1.0.1
+Requests==2.32.3

src/app/booklog-api.py

@@ -0,0 +1,59 @@
+import os
+import gradio as gr
+import requests
+from bs4 import BeautifulSoup
+
+class BooklogAPI:
+    def __init__(self, api_key):
+        self.base_url = "https://api.booklog.jp/v2/json"
+        self.api_key = api_key
+
+    def get_books(self, user_id, month):
+        endpoint = f"{self.base_url}/{user_id}?count=10000"
+        response = requests.get(endpoint)
+
+        if response.status_code == 200:
+            books = response.json()
+            # 特定の月に読んだ本の情報をフィルタリング
+            books_in_month = [book for book in books if book['read_date'][:7] == month]
+            return books_in_month
+        else:
+            print(f"Failed to fetch books. Status code: {response.status_code}")
+            return None
+
+    def get_book_details(self, isbn):
+        html = requests.get(f"https://booklog.jp/users/{self.api_key}/archives/1/{isbn}")
+        soup = BeautifulSoup(html.content, "html.parser")
+        register_date = soup.find(class_='read-day-status-area').find('span').text
+        amazon_link = soup.find(class_='itemInfoElm').find('a').get('href')
+        return register_date, amazon_link
+
+def booklog_app(month):
+    # BooklogAPIのインスタンスを作成
+    api_key = os.getenv("BOOKLOG_API_KEY", None)
+    booklog_api = BooklogAPI(api_key)
+
+    # ユーザーID
+    user_id = os.getenv("BOOKLOG_USER_ID", None)
+
+    # 特定の月に読んだ本の情報を取得
+    books_in_month = booklog_api.get_books(user_id, month)
+
+    if books_in_month:
+        book_details = []
+        for book in books_in_month:
+            isbn = book['isbn']
+            register_date, amazon_link = booklog_api.get_book_details(isbn)
+            book_details.append({
+                'title': book['title'],
+                'author': book['author'],
+                'register_date': register_date,
+                'amazon_link': amazon_link
+            })
+        return book_details
+    else:
+        return "No books found for the selected month."
+
+if __name__ == '__main__':
+    iface = gr.Interface(fn=booklog_app, inputs="text", outputs="table", title="Booklog Book Details")
+    iface.launch()

src/app/calil-api.py

@@ -0,0 +1,172 @@
+import os
+import time
+import requests
+import gradio as gr
+from dotenv import load_dotenv
+from pandas import json_normalize
+
+
+load_dotenv()
+
+readme = '''
+個別の本ページへのリンク：
+https://calil.jp/book/{ISBN10}
+図書館ページへのリンク：
+https://calil.jp/library/{libid}/{name}
+図書館ページへのリンク2（libkeyとシステムIDから飛ぶ場合）：
+https://calil.jp/library/search?s={systemid}&k={Libkey}
+'''
+
+class CalilAPI:
+    def __init__(self, app_key):
+        self.app_key = app_key
+        self.base_url = "https://api.calil.jp"
+
+    def search_library(self, pref, city):
+        url = f"{self.base_url}library"
+        params = {
+            'appkey': self.app_key,
+            'pref': pref,
+            'city': city,
+            'systemid': None, # Aomori_Pref
+            'geocode': None, # 136.7163027,35.390516
+            'format': 'json', # xml
+            'callback': '',
+            'limit': None
+        }
+        response = requests.get(url, params=params)
+        if response.status_code == 200:
+            try:
+                return response.json()
+            except requests.exceptions.JSONDecodeError:
+                return {"error": "Invalid JSON response"}
+        else:
+            response.raise_for_status()
+
+    def search_book(self, isbn, systemid):
+        url = f"{self.base_url}check"
+        params = {
+            'appkey': self.app_key,
+            'isbn': isbn,
+            'systemid': systemid,
+            'format': 'json',
+            'callback': 'no'
+        }
+        response = requests.get(url, params=params)
+        if response.status_code == 200:
+            try:
+                return response.json()
+                # return json_normalize(response.text, 'items')
+            except requests.exceptions.JSONDecodeError:
+                return {"error": "Invalid JSON response"}
+        else:
+            response.raise_for_status()
+
+class CalilOpenURLAPI(CalilAPI):
+    def __init__(self, app_key):
+        self.endp = '/openurl'
+
+    def search(self, query):
+        params = {
+            'rft.btitle': None,
+            'rft.title': None,
+            'rft.au': None,
+            'rft.aufirst': None,
+            'rft.aulast': None,
+            'rft.pub': None,
+        }
+
+class Parser:
+    @staticmethod
+    def handle_isbn(isbn):
+        return isbn.replace("-", "").strip()
+
+    @staticmethod
+    def parse_library_data(data):
+        if "error" in data:
+            return data["error"]
+        libraries = []
+        for library in data:
+            libraries.append({
+                'name': library.get('formal'),
+                'address': library.get('address'),
+                'systemid': library.get('systemid'),
+                'libkey': library.get('libkey')
+            })
+        return libraries
+
+    @staticmethod
+    def parse_book_data(data):
+        if "error" in data:
+            return data["error"]
+        books = []
+        for systemid, system_data in data.items():
+            if isinstance(system_data, dict) and system_data.get('status') == 'OK':
+                for libkey, lib_data in system_data['libkey'].items():
+                    books.append({
+                        'libkey': libkey,
+                        'status': lib_data
+                    })
+        return books
+
+def search_library_ui(pref, city):
+    app_key = os.getenv("CALIL_APP_KEY")
+    calil_api = CalilAPI(app_key)
+    parser = Parser()
+
+    library_data = calil_api.search_library(pref, city)
+    # parsed_data = parser.parse_library_data(library_data)
+    parsed_data = library_data
+    if isinstance(parsed_data, str):  # Error message case
+        return parsed_data
+    return parsed_data
+
+def search_book_ui(isbn, systemid):
+    app_key = os.getenv("CALIL_APP_KEY")
+    calil_api = CalilAPI(app_key)
+    parser = Parser()
+
+    cleaned_isbn = parser.handle_isbn(isbn)
+    book_data = calil_api.search_book(cleaned_isbn, systemid)
+    # parsed_data = parser.parse_book_data(book_data)
+    parsed_data = book_data
+    if isinstance(parsed_data, str):  # Error message case
+        return parsed_data
+    return parsed_data
+
+def create_ui():
+    with gr.Blocks() as demo:
+        gr.Markdown("# 図書館データベース検索 & 蔵書検索")
+
+        with gr.Tab("図書館検索"):
+            with gr.Row():
+                with gr.Column():
+                    pref_input = gr.Textbox(label="都道府県")
+                    city_input = gr.Textbox(label="市区町村")
+                    search_btn = gr.Button("検索")
+
+                    gr.Examples([['青森県', '青森市']], [pref_input, city_input])
+
+                with gr.Column(elem_classes='max-lines'):
+                    result_output = gr.JSON()
+                # result_output = gr.Dataframe(headers=["名前", "住所", "System ID", "Libkey"])
+                search_btn.click(fn=search_library_ui, inputs=[pref_input, city_input], outputs=result_output)
+
+        with gr.Tab("蔵書検索"):
+            with gr.Row():
+                with gr.Column():
+                    isbn_input = gr.Textbox(label="ISBN")
+                    systemid_input = gr.Textbox(label="System ID")
+                    search_btn = gr.Button("検索")
+
+                    gr.Examples([['4834000826', 'Aomori_Pref'], [['4834000826', 'Tokyo_Setagaya']]], [isbn_input, systemid_input])
+
+                with gr.Column(elem_classes='max-lines'):
+                    result_output = gr.JSON()
+                # result_output = gr.Dataframe(headers=["Libkey", "ステータス"])
+                search_btn.click(fn=search_book_ui, inputs=[isbn_input, systemid_input], outputs=result_output)
+
+    return demo
+
+if __name__ == "__main__":
+    create_ui().launch()

src/app/ndl-api.py

@@ -0,0 +1,86 @@
+
+class SruAPI:
+    def __init__(self):
+        self.endp = 'https://ndlsearch.ndl.go.jp/api/sru/'
+        self.params = {
+            'operation': '',  # searchRetrieve
+            'query': '', # CQL
+            'startRecord': '', # 1
+            'maximumRecord': '', # 200
+            'recordPacking': '', # string or xml
+            'recordSchema': '', # dc, dcndl
+        }
+
+class OpensearchAPI:
+    def __init__(self):
+        self.endp = 'https://ndlsearch.ndl.go.jp/api/opensearch/'
+        self.params = {
+            'title': '',
+            'creator': '',
+            'from': '',
+            'cnt': '', # 200
+            'idx': '', # 1
+        }
+
+class OpenURLAPI:
+    def __init__(self):
+        self.endp = ''
+
+class OAIPMH_API:
+    def __init__(self):
+        self.endp = '/'
+
+class ShoeiAPI:
+    def __init__(self):
+        self.endp = 'https://ndlsearch.ndl.go.jp/thumbnail/'
+
+    async def get_thumbnail(self, isbn, temp_dir):
+        async with aiohttp.ClientSession() as session:
+            try:
+                async with session.get(f'{self.endp}{isbn}') as response:
+                    if response.status == 200:
+                        image_data = await response.read()
+                        image_path = os.path.join(temp_dir, f'{isbn}.jpg')
+                        async with aiofiles.open(image_path, 'wb') as image_file:
+                            await image_file.write(image_data)
+                        return image_path
+                    else:
+                        return None
+            except aiohttp.ClientError as e:
+                print(f'Failed to retrieve thumbnail for {isbn}: {e}')
+                return None
+
+    async def export_html(self, book_data, template_path, temp_dir):
+        timestamp = datetime.now().strftime("%Y-%m-%d %H-%M-%S")
+        rows = ''
+        for i in range(0, len(book_data), 3):
+            rows += '<tr>'
+            for j in range(3):
+                if i + j < len(book_data):
+                    book = book_data[i + j]
+                    thumbnail = await self.get_thumbnail(book['isbn'], temp_dir)
+                    if thumbnail:
+                        rows += f"""
+                        <td align="center" valign="top" width="33.33%">
+                            <img src="{thumbnail}" style="width:100px;">
+                            <br /><sub><b>{book['isbn']}</b></sub></a><br /><description>{book['title']}</description><a href="https://booklog.jp/item/1/{book['isbn']}" >🔗</a>
+                        </td>
+                        """
+                    else:
+                        rows += '<td align="center" valign="top" width="33.33%"></td>'
+                else:
+                    rows += '<td align="center" valign="top" width="33.33%"></td>'
+            rows += '</tr>'
+
+        with open(template_path, 'r') as file:
+            html_template = file.read()
+
+        html_content = html_template.replace('$timestamp', timestamp).replace('$rows', rows)
+        return html_content
+
+    async def create_zip(self, temp_dir, output_zip):
+        with zipfile.ZipFile(output_zip, 'w') as zipf:
+            for root, _, files in os.walk(temp_dir):
+                for file in files:
+                    file_path = os.path.join(root, file)
+                    zipf.write(file_path, os.path.relpath(file_path, temp_dir))

src/app/run.py

@@ -10,61 +10,6 @@ import gradio as gr
 from datetime import datetime
 from isbnlib import to_isbn10
 
-class ShoeiAPI:
-    def __init__(self):
-        self.endp = 'https://ndlsearch.ndl.go.jp/thumbnail/'
-
-    async def get_thumbnail(self, isbn, temp_dir):
-        async with aiohttp.ClientSession() as session:
-            try:
-                async with session.get(f'{self.endp}{isbn}') as response:
-                    if response.status == 200:
-                        image_data = await response.read()
-                        image_path = os.path.join(temp_dir, f'{isbn}.jpg')
-                        async with aiofiles.open(image_path, 'wb') as image_file:
-                            await image_file.write(image_data)
-                        return image_path
-                    else:
-                        return None
-            except aiohttp.ClientError as e:
-                print(f'Failed to retrieve thumbnail for {isbn}: {e}')
-                return None
-
-    async def export_html(self, book_data, template_path, temp_dir):
-        timestamp = datetime.now().strftime("%Y-%m-%d %H-%M-%S")
-        rows = ''
-        for i in range(0, len(book_data), 3):
-            rows += '<tr>'
-            for j in range(3):
-                if i + j < len(book_data):
-                    book = book_data[i + j]
-                    thumbnail = await self.get_thumbnail(book['isbn'], temp_dir)
-                    if thumbnail:
-                        rows += f"""
-                        <td align="center" valign="top" width="33.33%">
-                            <img src="{thumbnail}" style="width:100px;">
-                            <br /><sub><b>{book['isbn']}</b></sub></a><br /><description>{book['title']}</description><a href="https://booklog.jp/item/1/{book['isbn']}" >🔗</a>
-                        </td>
-                        """
-                    else:
-                        rows += '<td align="center" valign="top" width="33.33%"></td>'
-                else:
-                    rows += '<td align="center" valign="top" width="33.33%"></td>'
-            rows += '</tr>'
-
-        with open(template_path, 'r') as file:
-            html_template = file.read()
-
-        html_content = html_template.replace('$timestamp', timestamp).replace('$rows', rows)
-        return html_content
-
-    async def create_zip(self, temp_dir, output_zip):
-        with zipfile.ZipFile(output_zip, 'w') as zipf:
-            for root, _, files in os.walk(temp_dir):
-                for file in files:
-                    file_path = os.path.join(root, file)
-                    zipf.write(file_path, os.path.relpath(file_path, temp_dir))
-
 
 class Parser:
     def __init__(self):

src/app/util.py

@@ -0,0 +1,45 @@
+
+
+class Interface:
+
+    def get_tempdir():
+        timestamp = int(time.time())
+        # timestamp = datetime.now().isoformat(timespec='auto')
+        temp_dir = tempfile.mkdtemp()
+        return timestamp, temp_dir
+
+    @staticmethod
+    def create_zip(filelist, tmp_fname, passwd=None):
+        if not filelist:
+            return None
+        try:
+            zip_name = os.path.abspath(tmp_fname)
+            with zipfile.ZipFile(zip_name, "w", compression=zipfile.ZIP_DEFLATED) as f:
+                for file in filelist:
+                    if os.path.isfile(file):
+                        f.write(file, os.path.relpath(file, os.path.dirname(filelist[0])))
+                    elif os.path.isdir(file):
+                        for root, dirs, files in os.walk(file):
+                            for filename in files:
+                                filepath = os.path.join(root, filename)
+                                f.write(filepath, os.path.relpath(filepath, os.path.dirname(filelist[0])))
+            if passwd:
+                zip_name_encrypted = zip_name + ".zip"
+                with zipfile.ZipFile(zip_name_encrypted, "w", compression=zipfile.ZIP_DEFLATED) as f:
+                    f.setpassword(passwd)
+                    f.write(zip_name, os.path.basename(zip_name))
+                os.remove(zip_name)
+                return zip_name_encrypted
+            else:
+                return zip_name
+        except Exception as e:
+            raise RuntimeError(f"Failed to create zip file: {str(e)}")
+
+    def read_csv(csv_file, fieldnames=None, encoding='utf-8'):
+        feeds = []
+        with open(csv_file, 'r', newline='', encoding=encoding) as csvfile:
+            reader = csv.DictReader(csvfile, fieldnames=fieldnames)
+            for row in reader:
+                feeds.append(row)
+        data = {"items": feeds}
+        return data