import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ExecutionContext;

// ItemReader核心接口
public interface ItemReader<T> {
    /**
     * 讀取一個(gè)數(shù)據(jù)項(xiàng)
     * 返回null表示讀取結(jié)束
     */
    T read() throws Exception;
}

// ItemStream接口用于管理讀取狀態(tài)
public interface ItemStream {
    /**
     * 打開(kāi)資源并初始化狀態(tài)
     */
    void open(ExecutionContext executionContext) throws ItemStreamException;
    
    /**
     * 更新執(zhí)行上下文中的狀態(tài)信息
     */
    void update(ExecutionContext executionContext) throws ItemStreamException;
    
    /**
     * 關(guān)閉資源并清理狀態(tài)
     */
    void close() throws ItemStreamException;
}

// CSV文件讀取器
@Bean
public FlatFileItemReader<Transaction> csvTransactionReader() {
    FlatFileItemReader<Transaction> reader = new FlatFileItemReader<>();
    reader.setResource(new FileSystemResource("data/transactions.csv"));
    reader.setLinesToSkip(1); // 跳過(guò)標(biāo)題行
    
    // 配置行映射器
    DefaultLineMapper<Transaction> lineMapper = new DefaultLineMapper<>();
    
    // 配置分隔符解析器
    DelimitedLineTokenizer tokenizer = new DelimitedLineTokenizer();
    tokenizer.setDelimiter(",");
    tokenizer.setNames("id", "date", "amount", "customerId", "type");
    lineMapper.setLineTokenizer(tokenizer);
    
    // 配置字段映射器
    BeanWrapperFieldSetMapper<Transaction> fieldSetMapper = new BeanWrapperFieldSetMapper<>();
    fieldSetMapper.setTargetType(Transaction.class);
    lineMapper.setFieldSetMapper(fieldSetMapper);
    
    reader.setLineMapper(lineMapper);
    reader.setName("transactionItemReader");
    
    return reader;
}

// JDBC分頁(yè)讀取器
@Bean
public JdbcPagingItemReader<Order> jdbcPagingReader(DataSource dataSource) throws Exception {
    JdbcPagingItemReader<Order> reader = new JdbcPagingItemReader<>();
    reader.setDataSource(dataSource);
    reader.setFetchSize(100); // 設(shè)置頁(yè)大小
    reader.setPageSize(100);
    
    // 配置查詢參數(shù)
    Map<String, Object> parameterValues = new HashMap<>();
    parameterValues.put("status", "PENDING");
    reader.setParameterValues(parameterValues);
    
    // 配置行映射器
    reader.setRowMapper(new OrderRowMapper());
    
    // 配置分頁(yè)查詢提供者
    SqlPagingQueryProviderFactoryBean queryProviderFactory = new SqlPagingQueryProviderFactoryBean();
    queryProviderFactory.setDataSource(dataSource);
    queryProviderFactory.setSelectClause("SELECT id, customer_id, order_date, total_amount, status");
    queryProviderFactory.setFromClause("FROM orders");
    queryProviderFactory.setWhereClause("WHERE status = :status");
    queryProviderFactory.setSortKey("id");
    
    reader.setQueryProvider(queryProviderFactory.getObject());
    
    return reader;
}

// 多資源文件讀取器
@Bean
public MultiResourceItemReader<Customer> multiResourceReader() throws IOException {
    MultiResourceItemReader<Customer> multiReader = new MultiResourceItemReader<>();
    
    // 獲取多個(gè)資源文件
    Resource[] resources = new PathMatchingResourcePatternResolver()
            .getResources("classpath:data/customers_*.csv");
    multiReader.setResources(resources);
    
    // 設(shè)置委托讀取器
    FlatFileItemReader<Customer> delegateReader = new FlatFileItemReader<>();
    // 配置委托讀取器...
    multiReader.setDelegate(delegateReader);
    
    return multiReader;
}

// REST API數(shù)據(jù)讀取器
@Component
public class RestApiItemReader<T> implements ItemReader<T>, ItemStream {
    
    private final RestTemplate restTemplate;
    private final String apiUrl;
    private final Class<T> targetType;
    
    private int currentPage = 0;
    private int pageSize = 100;
    private List<T> currentItems;
    private int currentIndex;
    private boolean exhausted = false;
    
    // 狀態(tài)保存鍵
    private static final String CURRENT_PAGE_KEY = "current.page";
    private String name;
    
    public RestApiItemReader(RestTemplate restTemplate, String apiUrl, Class<T> targetType) {
        this.restTemplate = restTemplate;
        this.apiUrl = apiUrl;
        this.targetType = targetType;
    }
    
    @Override
    public T read() throws Exception {
        // 如果當(dāng)前批次數(shù)據(jù)為空或已讀取完畢，加載下一批
        if (currentItems == null || currentIndex >= currentItems.size()) {
            if (exhausted) {
                return null; // 所有數(shù)據(jù)讀取完畢
            }
            
            fetchNextBatch();
            
            // 如果加載后列表為空，表示所有數(shù)據(jù)讀取完畢
            if (currentItems.isEmpty()) {
                exhausted = true;
                return null;
            }
            
            currentIndex = 0;
        }
        
        return currentItems.get(currentIndex++);
    }
    
    private void fetchNextBatch() {
        String url = apiUrl + "?page=" + currentPage + "&size=" + pageSize;
        ResponseEntity<ApiResponse<T>> response = restTemplate.getForEntity(
                url, (Class<ApiResponse<T>>) ApiResponse.class);
        
        ApiResponse<T> apiResponse = response.getBody();
        currentItems = apiResponse.getItems();
        exhausted = apiResponse.isLastPage();
        currentPage++;
    }
    
    // ItemStream接口實(shí)現(xiàn)省略...
}

// 數(shù)據(jù)分區(qū)策略
@Component
public class RangePartitioner implements Partitioner {
    
    @Override
    public Map<String, ExecutionContext> partition(int gridSize) {
        Map<String, ExecutionContext> partitions = new HashMap<>(gridSize);
        
        // 獲取數(shù)據(jù)總數(shù)
        long totalRecords = getTotalRecords();
        long targetSize = totalRecords / gridSize + 1;
        
        // 創(chuàng)建分區(qū)
        for (int i = 0; i < gridSize; i++) {
            ExecutionContext context = new ExecutionContext();
            
            long fromId = i * targetSize + 1;
            long toId = (i + 1) * targetSize;
            if (toId > totalRecords) {
                toId = totalRecords;
            }
            
            context.putLong("fromId", fromId);
            context.putLong("toId", toId);
            
            partitions.put("partition-" + i, context);
        }
        
        return partitions;
    }
    
    private long getTotalRecords() {
        // 獲取數(shù)據(jù)總數(shù)的實(shí)現(xiàn)
        return 1000000;
    }
}

// 配置錯(cuò)誤處理和狀態(tài)管理
@Bean
public Step robustStep(
        ItemReader<InputData> reader,
        ItemProcessor<InputData, OutputData> processor,
        ItemWriter<OutputData> writer,
        SkipPolicy skipPolicy) {
    
    return stepBuilderFactory.get("robustStep")
            .<InputData, OutputData>chunk(100)
            .reader(reader)
            .processor(processor)
            .writer(writer)
            // 配置錯(cuò)誤處理
            .faultTolerant()
            .skipPolicy(skipPolicy)
            .retryLimit(3)
            .retry(TransientDataAccessException.class)
            .noRetry(NonTransientDataAccessException.class)
            .listener(new ItemReadListener<InputData>() {
                @Override
                public void onReadError(Exception ex) {
                    // 記錄讀取錯(cuò)誤
                    logError("Read error", ex);
                }
            })
            .build();
}