基于条件创建DQ_Resolution_Date字段的Python技术求助

我在Python中编写脚本创建数据质量（DQ）问题历史表：执行多条SQL查询合并生成Today_DQ_Results，再拉取昨日结果Prior_Day_DQ_Results进行合并、去重。已成功用Date_Identified标记问题发现时间，但无法正确生成DQ_Resolution_Date字段——该字段需要标记那些昨日存在于Prior_Day_DQ_Results但今日未出现在Today_DQ_Results中的已解决DQ问题。

以下是我尝试实现该字段的代码：

# Define the queries
address_completeness = """SELECT ENTITY_ID, ADDRESS_LINE2 AS CONDITION_1, ADDRESS_LINE2 AS CONDITION_2, ENTITY_STATUS
FROM Table_1 
WHERE ADDRESS_LINE2 IS NULL"""

city_completeness = """SELECT ENTITY_ID, CITY_NAME AS CONDITION_1, CITY_NAME AS CONDITION_2, ENTITY_STATUS
FROM Table_1  
WHERE CITY_NAME IS NULL"""
    
Region_Subregion_Accuracy = """SELECT ENTITY_ID, DIVISION AS CONDITION_1, ENTITY_STATUS, SNDIVISION AS CONDITION_2
FROM Table_1
WHERE DIVISION <> SNDIVISION"""

# Execute the queries and save the results to dataframes
address_completeness_df = pd.read_sql(address_completeness, conn)
city_completeness_df = pd.read_sql(city_completeness, conn)
Region_Subregion_Accuracy_dr = pd.read_sql(Region_Subregion_Accuracy, conn)


# Add a new column called "DQ_Rule" to each dataframe and set its value accordingly
address_completeness_df['DQ_Rule'] = 'Address Completeness'
city_completeness_df['DQ_Rule'] = 'City Completeness'
Region_Subregion_Accuracy_dr['DQ_Rule'] = 'Region Subregion Accuracy'

# Combine the dataframes into a single dataframe
Today_DQ_Results = pd.concat([address_completeness_df, city_completeness_df, Region_Subregion_Accuracy_dr])

# Get current date in the format MMDDYY
current_date = datetime.datetime.now().strftime('%m%d%y')

# Add a new column called "Unique_ID" with a unique identifier for each row
Today_DQ_Results['Unique_ID'] = [secrets.token_hex(3) + '_' + current_date for _ in range(len(Today_DQ_Results))]

# Create a new column called "Date_Identified" and set its value to today's date
Today_DQ_Results['Date_Identified'] = pd.Timestamp.now().strftime('%Y-%m-%d')

# Rearrange the order of columns
column_order = ['Unique_ID', 'ENTITY_ID', 'ENTITY_STATUS', 'DQ_Rule', 'CONDITION_1', 'CONDITION_2', 'Date_Identified']
Today_DQ_Results = Today_DQ_Results.reindex(columns=column_order)

#### Pull Yesterday's DQ File - Update Daily ####
file_path = r'C:\\Users\\Ben\\Desktop\\DQ Python\\DQ_Summary_Report_6_12_23.xlsx'
Prior_Day_DQ_Results = pd.read_excel(file_path)

# combine the dataframes
Updated_DQ_Report = pd.concat([Prior_Day_DQ_Results, Today_DQ_Results], ignore_index=True)

# identify duplicate records
duplicates = Updated_DQ_Report.duplicated(subset=['ENTITY_ID', 'DQ_Rule', 'CONDITION_1', 'CONDITION_2'], keep=False)

# create a new 'DQ_Resolution_Date' column and set its value based on the presence of duplicates
today = datetime.date.today().strftime('%Y-%m-%d')
Updated_DQ_Report['DQ_Resolution_Date'] = np.where(duplicates, '', today)

# drop duplicates based on specified columns
Updated_DQ_Report = Updated_DQ_Report.drop_duplicates(subset=['ENTITY_ID', 'DQ_Rule', 'CONDITION_1', 'CONDITION_2'])

# Define a function to compare Date_Identified and DQ_Resolution_Date values
def update_resolution_date(row):
    if row['Date_Identified'] == row['DQ_Resolution_Date']:
        return ''
    else:
        return row['DQ_Resolution_Date']

# Apply the function to update DQ_Resolution_Date values
Updated_DQ_Report['DQ_Resolution_Date'] = Updated_DQ_Report.apply(update_resolution_date, axis=1)
Updated_DQ_Report.drop('Unnamed: 0', axis=1, inplace=True)

原代码逻辑错误：合并昨日和今日数据后，给非重复项标记今日为解决日期。但非重复项包含两部分：一是昨日存在今日消失的已解决问题，二是今日新发现的问题。这会导致今日新问题被错误标记为已解决，同时真正的已解决问题处理逻辑混乱。

核心思路是：

1. 精准识别仅存在于昨日结果中的问题（即今日已解决的问题），为其设置今日为DQ_Resolution_Date
2. 保留昨日存在今日仍存在的问题（未解决）的原有状态
3. 合并已解决问题、未解决问题和今日新发现问题

以下是修正后的关键代码（替换原代码中从Pull Yesterday's DQ File到最后的部分）：

#### Pull Yesterday's DQ File - Update Daily ####
file_path = r'C:\\Users\\Ben\\Desktop\\DQ Python\\DQ_Summary_Report_6_12_23.xlsx'
Prior_Day_DQ_Results = pd.read_excel(file_path)

# 获取今日日期，统一格式
today = pd.Timestamp.now().strftime('%Y-%m-%d')

# 定义唯一标识问题的组合键，确保匹配同一问题
match_keys = ['ENTITY_ID', 'DQ_Rule', 'CONDITION_1', 'CONDITION_2']

# 1. 筛选昨日存在但今日不存在的问题（已解决），标记今日为解决日期
resolved_mask = ~Prior_Day_DQ_Results.set_index(match_keys).index.isin(Today_DQ_Results.set_index(match_keys).index)
resolved_issues = Prior_Day_DQ_Results[resolved_mask].copy()
resolved_issues['DQ_Resolution_Date'] = today

# 2. 筛选昨日存在今日仍存在的问题（未解决），保留原有解决日期（若有）
unresolved_issues = Prior_Day_DQ_Results[~resolved_mask].copy()

# 3. 今日新发现的问题，解决日期留空（默认无该字段则自动添加）
if 'DQ_Resolution_Date' not in Today_DQ_Results.columns:
    Today_DQ_Results['DQ_Resolution_Date'] = ''

# 4. 合并所有数据：已解决问题 -> 未解决问题 -> 今日新问题
Updated_DQ_Report = pd.concat([resolved_issues, unresolved_issues, Today_DQ_Results], ignore_index=True)

# 清理不必要的列（忽略不存在的列避免报错）
Updated_DQ_Report.drop('Unnamed: 0', axis=1, inplace=True, errors='ignore')

• 组合键匹配：使用ENTITY_ID+DQ_Rule+CONDITION_1+CONDITION_2作为唯一标识，确保不会误判不同问题为同一问题。
• 数据分离处理：将昨日数据拆分为已解决和未解决两组，分别处理后再合并，逻辑清晰且避免错误。
• 容错处理：添加errors='ignore'避免因列不存在导致报错，同时为今日新数据默认添加空的解决日期字段，保证数据结构一致。

内容的提问来源于stack exchange，提问作者Benjamin