Python生成连续星期序列

Posted 2022-11-24 小基基o_O

文章目录

• 概述
• datetime模块
• 生成连续星期序列
• 生成日期维度表

概述

• 在商业数据分析中，会对销量按天、周、月、季、年汇总
  按星期聚合的格式可以是：2022年第01周、2022w51、202251…
• 在某些零售行业中，线下零售假日销量高于工作日
  因此，在按周汇总时，希望以星期一作为周的起点
• 对于需要跨年的场景，期望将
  以 年内第1个星期一 作为 年的第一周，此前归入上一年最后一周，例如下面

例如，2023-01-01归属到2022w52那周

datetime模块

from datetime import datetime

# 今天1024
d = datetime.strptime('2022-10-24', '%Y-%m-%d')
# 星期一
w = d.weekday()  # 0
# 今年的第几天
day_of_year = (d - datetime(year=d.year, month=1, day=1)).days + 1  # 297

生成连续星期序列

from datetime import datetime, timedelta
from math import ceil

START = '2022-12-25'
END = '2023-01-09'

def get_day_of_year(d: datetime) -> int:
    return (d - datetime(year=d.year, month=1, day=1)).days + 1

def get_week_of_year(d: datetime) -> int:
    day_of_year = get_day_of_year(d)
    return ceil((day_of_year - d.weekday()) / 7)

def get_weekday_cn(d: datetime) -> str:
    return ['周一', '周二', '周三', '周四', '周五', '周六', '周日'][d.weekday()]

def get_year_week(d: datetime) -> (int, int, str):
    day_of_year = get_day_of_year(d)
    week_of_year = get_week_of_year(d)
    if week_of_year == 0:
        y = d.year - 1
        yw = '%04dw%02d' % (y, get_week_of_year(datetime(year=y, month=12, day=31)))
    else:
        y = d.year
        yw = '%04dw%02d' % (y, week_of_year)
    return day_of_year, week_of_year, yw

def generate_date():
    start = datetime.strptime(START, '%Y-%m-%d')
    end = datetime.strptime(END, '%Y-%m-%d')
    for i in range((end - start).days + 1):
        d = start + timedelta(i)
        ymd = d.strftime('%Y-%m-%d')
        day_of_year, week_of_year, yw = get_year_week(d)
        weekday_cn = get_weekday_cn(d)
        print(ymd, day_of_year, week_of_year, yw, weekday_cn, sep='\\t')

if __name__ == '__main__':
    generate_date()

生成日期维度表

2022年，含节假日

from datetime import datetime, timedelta
from math import ceil

START = '2022-01-01'
END = '2022-12-31'
HOLIDAY = 
    '2022-01-01': ('元旦', False), '2022-01-02': ('元旦', False),
    '2022-01-03': ('元旦', False), '2022-01-29': ('补班', True),
    '2022-01-30': ('补班', True), '2022-01-31': ('春节', False),
    '2022-02-01': ('春节', False), '2022-02-02': ('春节', False),
    '2022-02-03': ('春节', False), '2022-02-04': ('春节', False),
    '2022-02-05': ('春节', False), '2022-02-06': ('春节', False),
    '2022-04-02': ('补班', True), '2022-04-03': ('清明节', False),
    '2022-04-04': ('清明节', False), '2022-04-05': ('清明节', False),
    '2022-04-24': ('补班', True), '2022-04-30': ('劳动节', False),
    '2022-05-01': ('劳动节', False), '2022-05-02': ('劳动节', False),
    '2022-05-03': ('劳动节', False), '2022-05-04': ('劳动节', False),
    '2022-05-07': ('补班', True), '2022-06-03': ('端午节', False),
    '2022-06-04': ('端午节', False), '2022-06-05': ('端午节', False),
    '2022-09-11': ('中秋节', False), '2022-09-12': ('中秋节', False),
    '2022-09-10': ('中秋节', False), '2022-10-01': ('国庆节', False),
    '2022-10-02': ('国庆节', False), '2022-10-03': ('国庆节', False),
    '2022-10-04': ('国庆节', False), '2022-10-05': ('国庆节', False),
    '2022-10-06': ('国庆节', False), '2022-10-07': ('国庆节', False),
    '2022-10-08': ('补班', True), '2022-10-09': ('补班', True),


def get_day_of_year(d: datetime) -> int:
    return (d - datetime(year=d.year, month=1, day=1)).days + 1

def get_week_of_year(d: datetime) -> int:
    day_of_year = get_day_of_year(d)
    return ceil((day_of_year - d.weekday()) / 7)

def get_weekday_cn(d: datetime) -> str:
    return ['周一', '周二', '周三', '周四', '周五', '周六', '周日'][d.weekday()]

def get_year_week(d: datetime) -> (int, int, str):
    day_of_year = get_day_of_year(d)
    week_of_year = get_week_of_year(d)
    if week_of_year == 0:
        y = d.year - 1
        yw = '%04dw%02d' % (y, get_week_of_year(datetime(year=y, month=12, day=31)))
    else:
        y = d.year
        yw = '%04dw%02d' % (y, week_of_year)
    return day_of_year, week_of_year, yw

def generate_date():
    start = datetime.strptime(START, '%Y-%m-%d')
    end = datetime.strptime(END, '%Y-%m-%d')
    for i in range((end - start).days + 1):
        d = start + timedelta(i)
        ymd = d.strftime('%Y-%m-%d')
        ym = d.strftime('%Y-%m')
        yq = '%d%s%d' % (d.year, 'q', ceil(d.month / 3))
        day_of_year, week_of_year, yw = get_year_week(d)
        weekday_cn = get_weekday_cn(d)
        holiday, is_workday = HOLIDAY.get(ymd, (
            '休息日' if d.weekday() > 4 else '工作日',
            d.weekday() < 5))
        print(ymd, ym, yq, yw, weekday_cn, day_of_year, week_of_year, holiday, is_workday, sep=',')

if __name__ == '__main__':
    generate_date()

HIVE建表

CREATE TABLE dim_date (
  ymd            STRING   COMMENT '日期',
  ym             STRING   COMMENT '年月',
  yq             STRING   COMMENT '季度',
  yw             STRING   COMMENT '星期，以周一为起点',
  weekday_cn     STRING   COMMENT '星期',
  day_of_year    SMALLINT COMMENT '年内第几天',
  week_of_year   TINYINT  COMMENT '年内第几周，以周一为起点',
  holiday        STRING   COMMENT '节假日',
  is_workday     BOOLEAN  COMMENT 'true=工作日、false=休息日'
) COMMENT '日期维度表'
ROW FORMAT DELIMITED FIELDS TERMINATED BY ',';