format code

course_list_info_parser.py

@@ -0,0 +1,137 @@
+# -*- coding: utf-8 -*-
+import configparser
+import sqlite3
+
+import psycopg2
+import requests
+
+import json
+from headers import headers
+from logging_config import setup_logging
+
+logger = setup_logging()
+
+# 读取配置文件
+config = configparser.ConfigParser()
+config.read('config.ini')
+authorization_token = config['DEFAULT']['authorization_token']
+limit = config['DEFAULT']['limit']
+offset = config['DEFAULT']['offset']
+sort = config['DEFAULT']['sort']
+
+# PostgreSQL 数据库配置
+dbname = config['POSTGRES']['dbname']
+user = config['POSTGRES']['user']
+password = config['POSTGRES']['password']
+host = config['POSTGRES']['host']
+port = config['POSTGRES']['port']
+
+headers = headers
+headers['authorization'] = f'Bearer {authorization_token}'
+
+# response = requests.get('https://trumacurl', headers=headers)
+
+params = {
+    'limit': limit,
+    'offset': offset,
+    # 'sort': 'oldest-first',
+    'sort': sort,
+}
+
+
+def get_list():
+    response = requests.get('https://bandu-api.songy.info/v2/communities/f0495084-4c6f-4f35-b4d5-2068641a53a1/courses',
+                            params=params, headers=headers)
+
+    # 假设response的内容就是你提供的course_list.json
+    course_list_json = response.text
+
+    logger.info(course_list_json)
+
+    json_data = json.loads(course_list_json)
+    course_item = json_data['data']['items']
+    for item in course_item[:10]:
+        logger.info(item['id'], item['title'], item['description'])
+
+    # 解析JSON数据
+    return json.loads(course_list_json)
+
+
+def insert_sqlit(course_list_data):
+    # 连接到SQLite数据库
+    # 如果文件不存在，会自动在当前目录创建:
+    conn = sqlite3.connect('courses.db')
+    cursor = conn.cursor()
+
+    # 创建一个表，如果它不存在:
+    cursor.execute('''
+    CREATE TABLE IF NOT EXISTS courses (
+        id INTEGER PRIMARY KEY,
+        title TEXT,
+        description TEXT
+    )
+    ''')
+
+    # 插入数据
+    for item in course_list_data['data']['items']:
+        cursor.execute('''
+        INSERT INTO courses (id, title, description)
+        VALUES (?, ?, ?)
+        ON CONFLICT(id) DO NOTHING
+    ''', (item['id'], item['title'], item['description']))
+
+    # 提交事务:
+    conn.commit()
+
+    # 关闭Cursor和Connection:
+    cursor.close()
+    conn.close()
+
+
+def insert_pgsql(course_list_data):
+    # 连接到PostgreSQL数据库
+    conn = psycopg2.connect(
+        dbname=dbname,
+        user=user,
+        password=password,
+        host=host,
+        port=port
+    )
+    cursor = conn.cursor()
+
+    # 创建一个表，如果它不存在:
+    cursor.execute('''
+        CREATE TABLE IF NOT EXISTS courses (
+            id INTEGER PRIMARY KEY,
+            title TEXT,
+            description TEXT
+        )
+        ''')
+
+    # 插入数据
+    for item in course_list_data['data']['items']:
+        try:
+            cursor.execute('''
+                INSERT INTO courses (id, title, description)
+                VALUES (%s, %s, %s)
+                ON CONFLICT (id) DO NOTHING
+                ''', (item['id'], item['title'], item['description']))
+        except psycopg2.IntegrityError:
+            # 如果发生唯一键冲突，忽略该插入
+            conn.rollback()
+        else:
+            conn.commit()
+
+    # 关闭Cursor和Connection:
+    cursor.close()
+    conn.close()
+
+
+def insert_data():
+    course_list_data = get_list()
+    insert_sqlit(course_list_data)
+    insert_pgsql(course_list_data)
+
+
+if __name__ == '__main__':
+    insert_data()