138 lines
3.4 KiB
Python
Executable File
138 lines
3.4 KiB
Python
Executable File
# -*- 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("ID: %s, Title: %s, Description: %s", 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()
|