python 写入一年的时间到数据库中_insertdate在python中的用法-CSDN博客

本文链接：https://blog.csdn.net/MR_comentropy/article/details/114050223

该博客介绍了如何使用Python配合Oracle数据库创建和填充交易日表。首先创建了两个表TEMPDATA_AA和T_XTJYR，并定义了字段和存储参数。接着，通过获取日历数据和调用百度接口获取节假日信息，判断每个日期是否为工作日。最后，将数据写入临时表和目标表，用于记录每年的交易日与非交易日情况。

摘要生成于 C知道，由 DeepSeek-R1 满血版支持，前往体验 >

python写入一年时间到系统交易日表，区分交易日非交易日，效果如下图

在这里插入图片描述

1.需要用到的临时表和目标表

1.1 创建临时表

-- Create table
create table TEMPDATA_AA
(
  c1  VARCHAR2(30),
  c2  VARCHAR2(30),
  c3  VARCHAR2(30),
  c4  VARCHAR2(30),
  c5  VARCHAR2(30),
  c6  VARCHAR2(30),
  c7  VARCHAR2(30),
  c8  VARCHAR2(30),
  c9  VARCHAR2(30),
  c10 VARCHAR2(30),
  n1  NUMBER(9),
  n2  NUMBER(9),
  n3  NUMBER(9),
  n4  NUMBER(9),
  n5  NUMBER(9),
  n6  NUMBER(9),
  n7  NUMBER(9),
  n8  NUMBER(9),
  n9  NUMBER(9),
  n10 NUMBER(9)
)
tablespace USERS
  pctfree 10
  initrans 1
  maxtrans 255
  storage
  (
    initial 64K
    next 1M
    minextents 1
    maxextents unlimited
  );

1.2 创建目标表

-- Create table
create table T_XTJYR
(
  naturalday   NUMBER(8),
  tradeday     NUMBER(8),
  year         NUMBER(4),
  quarter      NUMBER(1),
  month        NUMBER(2),
  week         NUMBER(1),
  year_month   NUMBER(6),
  year_weeknum NUMBER(6),
  weeknum      NUMBER(2),
  is_workday   NUMBER(1)
)
tablespace USERS
  pctfree 10
  initrans 1
  maxtrans 255
  storage
  (
    initial 64K
    next 1M
    minextents 1
    maxextents unlimited
  );
-- Create/Recreate primary, unique and foreign key constraints 
alter table T_XTJYR
  add unique (NATURALDAY)
  using index 
  tablespace USERS
  pctfree 10
  initrans 2
  maxtrans 255
  storage
  (
    initial 64K
    next 1M
    minextents 1
    maxextents unlimited
  );

2.准备和插入数据

# oracle 数据库测试连接
import cx_Oracle
import pandas as pd
con = cx_Oracle.connect('scott/123@localhost:1521/orcl',encoding='utf-8')
sql = ''' SELECT * FROM tempdata_aa a '''
cur = con.cursor()
cur.execute(sql)
df = pd.read_sql(sql,con)  # 只能读不能写数据
print(df)

# 多维列表转换成一维列表，后面提取每年的数据时会用到
from calendar import Calendar
def array_to_list(array):
    if not isinstance(array, (list, )):
        return [array]
    else:
        list_y = []
        for item in array:
            list_y += array_to_list(item)
    # 转换去重 
    return list(set(list_y))

if __name__ == '__main__':
#     a = [[[1,2],3],[4,[5,6]],[6,8,9]]
#     print(array_to_list(a))
    b =  Calendar(firstweekday=0).yeardatescalendar(2021,12)
    c = array_to_list(b)
    print(c,len(c))
#     d = list(map(lambda n: int(n.strftime('%Y%m%d')),c))
#     print(d)

# 获取某年的节假日列表，编写判断否是节假日函数
import requests
import json


# 从百度的php接口中获取到数据
def catch_url_from_baidu(calcultaion_year, month):
    headers = {
        "Content-Type": "application/json;charset=UTF-8"
    }
    param = {
        "query": calcultaion_year + "年" + month + "月",
        "resource_id": "39043",
        "t": "1604395059555",
        "ie": "utf8",
        "oe": "gbk",
        "format": "json",
        "tn": "wisetpl",
        "cb": ""
    }
    # 抓取位置：百度搜索框搜索日历，上面的日历的接口，可以在页面上进行核对
    r = requests.get(url="https://sp0.baidu.com/8aQDcjqpAAV3otqbppnN2DJv/api.php",
                     headers=headers, params=param).text
    month_data = json.loads(r)["data"][0]["almanac"]
    not_work_day = []
    for one in month_data:
        # 先转换成int类型的
        date_str=one["year"] + "-" + one["month"] + "-" + one["day"]
        c = int(dt.strptime(date_str,'%Y-%m-%d').strftime('%Y%m%d'))
        # 开始判断节假日类型
        if (one["cnDay"] == '日' or one["cnDay"] == '六'):
            if ('status' in one):
                if (one["status"] == "2"):
                    # status为2的时候表示周末的工作日，比如10月10日。即百度工具左上角显示“班”的日期
                    continue
                else:
                    # 普通周末时间
                    not_work_day.append(c)
                    continue
            else:
                # 普通周末时间。（接口中，如果左上角没有特殊表示，则不会返回status）
                not_work_day.append(c)
                continue
        if ('status' in one and one["status"] == "1"):
            # status为1的时候表示休息日，比如10月1日。即百度工具左上角显示“休”的日期
            not_work_day.append(c)
    return not_work_day


# 输出格式，可以修改成insert语句进行输出
def print_info(calcultaion_year):
    # 此处只能算当年之前的，因为国务院是每年12月份才会发布第二年的放假计划，所以此接口对于下一年的统计是错的。
    # eg：2020年11月4日，国务院没有发布21年的放假计划，那查询2021年元旦的时候，元旦那天不显示休息
    calculation_month = ["2", "5", "8", "11"]
    # 因该接口传入的时间，查询了前一个月，当前月和后一个月的数据，所以只需要2、5、8、11即可全部获取到。比如查询5月份，则会查询4,5,6月分的数据
    list_not_workday=[]
    for one_month in calculation_month:
        a = catch_url_from_baidu(calcultaion_year, one_month)
        list_not_workday.append(a)
    list_not_workday = array_to_list(list_not_workday)
    return list_not_workday

# 判断是否是交易日函数
def is_workday(num1):
    if num1 in list_not_workday:
        return 0
    else:
        return 1

if __name__ == '__main__':
    calcultaion_year = "2020"
    print_info(calcultaion_year)
    print(list_not_workday)
    print(len(list_not_workday))

# 数据写入临时表
from calendar import Calendar
def insert_date(year):
    """ 先写入一年的自然日到临时表 tempdata_aa 中 """
    date_array = Calendar(firstweekday=0).yeardatescalendar(year,12)
    date_list = array_to_list(date_array)
    con = cx_Oracle.connect('scott/123@localhost:1521/orcl',encoding='utf-8')
#     sql = ''' DELETE FROM tempdata_aa WHERE 1=1 '''
#     con.cursor().execute(sql)
#     con.commit()
    j = 0
    for i in date_list:
        if int(i.year) == year: # 过滤掉年头和年尾的非本年日期
            i = int(i.strftime('%Y%m%d'))
            # 是交易日，即为i，
            # 是节假日，执行小于自然日且为交易日的最大日期
            is_work = is_workday(i) # 0是节假日，1是交易日
            try:
                sql =f'''INSERT INTO  tempdata_aa(n1,n2)  VALUES({i},{is_work})'''
                con.cursor().execute(sql)
                con.commit()
                j += 1
                print('第{}次插入数据成功'.format(j))
            except Exception as e:
                print('第{}次数据插入失败：{}'.format(j, e))
                continue 
    # 关闭连接
    if 'con' in dir():
        con.close()

if __name__ == '__main__':
    year = 2021
    insert_date(year)                

# 数据准备完成，开始向目标表中插入数据
# 如果该年的数据已经插入，则删除该数据或跳出程序
try:
    con = cx_Oracle.connect('scott/123@localhost:1521/orcl',encoding='utf-8')
    sql =f''' INSERT INTO t_xtjyr
                      (naturalday,
                       tradeday,
                       YEAR,
                       quarter,
                       MONTH,
                       week,
                       year_month,
                       year_weeknum,
                       weeknum)
                      SELECT to_char(n1, 'yyyymmdd') AS naturalday,
                             tradeday,
                             to_char(n1, 'yyyy') AS YEAR,
                             to_char(n1, 'q') AS quarter,
                             to_char(n1, 'mm') AS MONTH,
                             to_char(n1, 'd') - 1 AS week, --0 为周末
                             to_char(n1, 'yyyymm') AS year_month,
                             to_char(n1, 'yyyyww') AS year_weeknum,
                             to_char(n1, 'ww') AS weeknum
                        FROM (SELECT to_date(a2.n1, 'yyyymmdd') AS n1,
                                     MAX(a1.n1) AS tradeday,
                                     a2.n2
                                FROM tempdata_aa a2
                                LEFT JOIN tempdata_aa a1
                                  ON a1.n1 <= a2.n1
                                 AND a1.n2 = 1
                               GROUP BY a2.n1, a2.n2
                               ORDER BY a2.n1)
--注意该方法在年初的几天假里，获取去去年的交易日，如果去年没有数据就会出现null值 '''
    con.cursor().execute(sql)
    con.commit()
    
except Exception as e:
    print('数据插入失败：{}'.format(e))
finally:
    if 'con' in dir():
        con.close()    


# 目标表主数据完成，开始写接口 有时间写吧



注意：
1.如果是第一次时开始使用节假日，一定要先把 list_not_workday列表先插入所有的数据，然后再检查临时表 tempdata_aa
2.表t_test 提供了2000年——2021的数据,也可以直接使用，当然后续每年12月份更新次年的数据就要使用上面的逻辑