Spaces:
Running
Running
| import os | |
| import pandas as pd | |
| import re | |
| import ast | |
| import logging | |
| from datetime import datetime | |
| # Configure logging | |
| logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') | |
| def clean_data(raw_csv_path, output_csv_path): | |
| if not os.path.exists(raw_csv_path): | |
| logging.error(f"Raw CSV not found at {raw_csv_path}") | |
| return | |
| df_raw = pd.read_csv(raw_csv_path) | |
| cleaned_records = [] | |
| for _, row in df_raw.iterrows(): | |
| commodity = row['Commodity'] | |
| raw_line = row['Raw_Line'] | |
| source_file = row['Source_File'] | |
| # Extract date from source file, e.g., price_report_20260420_e.pdf or price_report_20241114.pdf | |
| date_match = re.search(r'_(\d{8})', source_file) | |
| if date_match: | |
| date_str = date_match.group(1) | |
| date_obj = datetime.strptime(date_str, "%Y%m%d").date() | |
| else: | |
| logging.warning(f"Could not extract date from {source_file}, skipping.") | |
| continue | |
| # Clean the raw line to fix spacing in numbers (e.g., "1 29.00" -> "129.00") | |
| # Also handle space before comma (e.g., "1 ,010.00" -> "1,010.00") | |
| line = str(raw_line).replace(' ,', ',') | |
| line = re.sub(r'(\d)\s+(?=\d)', r'\1', line) | |
| # Re-extract prices using the cleaner line | |
| prices_str = re.findall(r'\d{1,3}(?:,\d{3})*(?:\.\d+)?', line) | |
| valid_prices = [] | |
| for p in prices_str: | |
| try: | |
| # Remove commas and convert to float | |
| val = float(p.replace(',', '')) | |
| # Filter out obvious non-prices like small single digits if they happen to be misparsed | |
| if val > 0: | |
| valid_prices.append(val) | |
| except ValueError: | |
| continue | |
| if valid_prices: | |
| # Average the prices found for that commodity on that day | |
| avg_price = sum(valid_prices) / len(valid_prices) | |
| cleaned_records.append({ | |
| "Date": date_obj, | |
| "Commodity": commodity, | |
| "Price": round(avg_price, 2) | |
| }) | |
| if cleaned_records: | |
| df_clean = pd.DataFrame(cleaned_records) | |
| # Average again in case there are multiple rows for the same commodity on the same day (e.g., Potato Local vs Potato Imp) | |
| df_clean = df_clean.groupby(['Date', 'Commodity'])['Price'].mean().reset_index() | |
| df_clean['Date'] = pd.to_datetime(df_clean['Date']) | |
| df_clean.sort_values(by=['Date', 'Commodity'], inplace=True) | |
| os.makedirs(os.path.dirname(output_csv_path), exist_ok=True) | |
| df_clean.to_csv(output_csv_path, index=False) | |
| logging.info(f"Cleaned data saved to {output_csv_path} with {len(df_clean)} records.") | |
| print(df_clean.head()) | |
| else: | |
| logging.warning("No valid prices could be extracted.") | |
| if __name__ == "__main__": | |
| raw_path = "data/processed/parsed_prices_raw.csv" | |
| output_path = "data/processed/clean_prices.csv" | |
| # Adjust paths if script is run from src/ingestion instead of project root | |
| if not os.path.exists(raw_path): | |
| raw_path = "../../data/processed/parsed_prices_raw.csv" | |
| output_path = "../../data/processed/clean_prices.csv" | |
| clean_data(raw_path, output_path) | |