This commit is contained in:
2026-10-14 00:07:12 +08:00
parent 3d3507a770
commit d029f9a19e
16 changed files with 655 additions and 29 deletions
@@ -0,0 +1,30 @@
package com.baiyue.lastleg.selfoperated.analytics.controller;
import com.baiyue.common.core.api.ApiResponse;
import com.baiyue.common.security.AuthenticatedUser;
import com.baiyue.lastleg.selfoperated.analytics.query.*;
import io.swagger.v3.oas.annotations.Operation;
import org.springframework.http.HttpStatus;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.security.core.annotation.AuthenticationPrincipal;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.server.ResponseStatusException;
import java.util.Map;
@RestController
@RequestMapping("/api/platform/v1/last-leg/self-operated-analytics")
@PreAuthorize("hasAuthority('platform:self-operated-analytics:list')")
public class PlatformInventoryTurnoverController {
private final InventoryTurnoverQueryService service;
public PlatformInventoryTurnoverController(InventoryTurnoverQueryService service){this.service=service;}
@GetMapping("/inventory-turnover")
@Operation(summary="固定版本的7/15/30日SKU与货主库存周转")
public ApiResponse<InventoryTurnoverResponses.Response> query(@AuthenticationPrincipal AuthenticatedUser user,
@ModelAttribute InventoryTurnoverRequest request,@RequestParam Map<String,String> parameters) {
if(user==null||user.tenantId()==null)throw new ResponseStatusException(HttpStatus.UNAUTHORIZED);
if(!"PLATFORM".equals(user.appCode())||user.tenantId()!=0L||user.impersonated()||!user.permissions().contains(PlatformAnalyticsController.READ_PERMISSION))
throw new ResponseStatusException(HttpStatus.FORBIDDEN,"平台自营分析只读权限不足");
InventoryTurnoverRequest.validateParameters(parameters.keySet());
return ApiResponse.success(service.query(request));
}
}
@@ -0,0 +1,71 @@
package com.baiyue.lastleg.selfoperated.analytics.query;
import com.baiyue.common.core.exception.BizException;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.time.*;
import java.time.temporal.ChronoUnit;
import java.util.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import static com.baiyue.lastleg.selfoperated.analytics.query.InventoryTurnoverResponses.*;
/** Tokens carry evidence, not authority: all versions and missing dates are reconstructed at asOf. */
@Component
public class InventoryTurnoverContext {
static final ZoneId ZONE=ZoneId.of("Asia/Shanghai");
static final String RULE="DAILY_SCHEDULED_FIRST_V1";
private final AnalyticsReadMapper analytics;
private final InventoryTurnoverReadMapper mapper;
private final ObjectMapper json;
private final Clock clock;
@Autowired public InventoryTurnoverContext(AnalyticsReadMapper analytics,InventoryTurnoverReadMapper mapper,ObjectMapper json) {
this(analytics,mapper,json,Clock.systemUTC());
}
InventoryTurnoverContext(AnalyticsReadMapper analytics,InventoryTurnoverReadMapper mapper,ObjectMapper json,Clock clock) {
this.analytics=analytics;this.mapper=mapper;this.json=json;this.clock=clock;
}
public record Evidence(String account,String warehouse,LocalDate end,int days,Instant asOf,
Long salesRun,Long inventoryRun,String salesHash,String inventoryHash,boolean rebuild,
List<Sample> samples,LocalDate accumulationStartedOn) {}
record Frozen(Context response,AnalyticsQueryService.Snapshot snapshot) {}
Frozen resolve(InventoryTurnoverRequest q) {
Instant now=clock.instant(); q.validate(now.atZone(ZONE).toLocalDate());
Evidence supplied=q.snapshotContext()==null?null:decode(q.snapshotContext());
Instant asOf=supplied==null?now.truncatedTo(ChronoUnit.MICROS):supplied.asOf();
if(asOf==null||asOf.isAfter(now)||!now.isBefore(asOf.plusSeconds(1800))) throw refresh();
LocalDate end=q.end(asOf.atZone(ZONE).toLocalDate());
if(supplied!=null&&(!Objects.equals(q.sourceAccount(),supplied.account())||!Objects.equals(q.warehouseCode(),supplied.warehouse())
||!end.equals(supplied.end())||q.days()!=supplied.days())) throw refresh();
var scopes=analytics.scopes(q.sourceAccount(),q.warehouseCode());
var sales=version(scopes,"OUTBOUND",asOf); var stock=version(scopes,"INVENTORY",asOf);
var aq=new AnalyticsQuery(q.sourceAccount(),q.warehouseCode(),null,null,"CUSTOM",end.minusDays(29),end,"DAY",null,1,20,
sales==null?0L:sales.id(),stock==null?0L:stock.id(),sales==null?null:sales.scopeHash(),stock==null?null:stock.scopeHash(),asOf);
// Staleness is evaluated at queryAsOf, so pagination does not age a fixed snapshot midway.
var snapshot=new AnalyticsQueryService(analytics,json,Clock.fixed(asOf,ZoneOffset.UTC)).context(aq,true,false,false,null);
var mapping=new TreeMap<LocalDate,Long>();
for(var sample:mapper.samples(q.sourceAccount(),q.warehouseCode(),utc(asOf),end.minusDays(29),end.plusDays(1))) {
if(sample.sampleDate()==null||sample.runId()==null||mapping.put(sample.sampleDate(),sample.runId())!=null) throw refresh();
}
List<Sample> samples=new ArrayList<>();
for(LocalDate date=end.minusDays(29);!date.isAfter(end);date=date.plusDays(1)) samples.add(new Sample(date,mapping.get(date)));
var began=mapper.accumulationStartedOn(q.sourceAccount(),q.warehouseCode(),utc(asOf));
var c=snapshot.context();
var evidence=new Evidence(q.sourceAccount(),q.warehouseCode(),end,q.days(),asOf,c.salesRunId(),c.inventoryRunId(),
c.salesScopeHash(),c.inventoryScopeHash(),c.rebuildRequired(),List.copyOf(samples),began);
if(supplied!=null&&!supplied.equals(evidence)) throw refresh();
return new Frozen(new Context(encode(evidence),q.sourceAccount(),q.warehouseCode(),end,q.days(),asOf,c.queryExpiresAt(),c,
c.inventoryObservedFrom(),c.inventoryObservedUntil(),RULE,List.copyOf(samples),samples.stream().filter(s->s.runId()==null).map(Sample::date).toList(),began),snapshot);
}
private AnalyticsReadRows.Run version(List<AnalyticsReadRows.Scope> scopes,String family,Instant asOf) {
var scope=scopes.stream().filter(s->family.equals(s.dataFamily())).findFirst().orElse(null);
return scope==null?null:mapper.versionAt(scope.id(),scope.scopeHash(),utc(asOf));
}
private Evidence decode(String token) {
try {var value=json.readValue(Base64.getUrlDecoder().decode(token),Evidence.class);if(value==null)throw refresh();return value;} catch(Exception e){throw refresh();}
}
private String encode(Evidence evidence) {
try{return Base64.getUrlEncoder().withoutPadding().encodeToString(json.writeValueAsBytes(evidence));}catch(Exception e){throw new IllegalStateException("Cannot encode turnover context",e);}
}
private static LocalDateTime utc(Instant value){return LocalDateTime.ofInstant(value,ZoneOffset.UTC);}
private static BizException refresh(){return new BizException("库存周转上下文无效或已过期,请整体刷新");}
}
@@ -0,0 +1,78 @@
package com.baiyue.lastleg.selfoperated.analytics.query;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.LocalDate;
import java.util.*;
import static com.baiyue.lastleg.selfoperated.analytics.query.InventoryTurnoverResponses.*;
/** Maps separately qualified quantities; a missing value never becomes a formal zero. */
public final class InventoryTurnoverPolicy {
private InventoryTurnoverPolicy() {}
static Map<String,Object> normalized(Map<String,Object> source) {
Map<String,Object> result=new HashMap<>();
if(source!=null)source.forEach((key,value)->result.put(key.toLowerCase(Locale.ROOT),value));
return result;
}
static BigDecimal decimal(Map<String,Object> row,String key) {
Object value=row.get(key);return value==null?null:new BigDecimal(value.toString());
}
static int count(Map<String,Object> row,String key){BigDecimal value=decimal(row,key);return value==null?0:value.intValueExact();}
static String text(Map<String,Object> row,String key){Object value=row.get(key);return value==null?null:value.toString();}
static BigDecimal divide(BigDecimal a,int n){return a==null?null:a.divide(BigDecimal.valueOf(n),6,RoundingMode.HALF_UP);}
private static BigDecimal scale(BigDecimal value){return value==null?null:value.setScale(6,RoundingMode.HALF_UP);}
static List<Window> windows(Map<String,Object> row,LocalDate until) {
List<Window> result=new ArrayList<>();
for(int n:List.of(7,15,30)) {
String key="w"+n+"_";var formal=decimal(row,key+"outbound");var observed=decimal(row,key+"observed");
int complete=count(row,key+"complete"),partial=count(row,key+"partial");
result.add(new Window(n,until.minusDays(n),until,formal,observed,divide(formal,n),complete,partial,
n-complete-partial,formal!=null?"COMPLETE":observed!=null?"PARTIAL":"UNKNOWN",
formal!=null?List.of():List.of("SALES_WINDOW_INCOMPLETE")));
}
return List.copyOf(result);
}
static Total total(Map<String,Object> row,String field,boolean owner) {
var value=decimal(row,field);int unknown=owner?count(row,field+"_unknown"):value==null?1:0;
return new Total(value,owner?decimal(row,field+"_known"):value,unknown,unknown==0?List.of():List.of("INVENTORY_UNKNOWN_OR_STALE"));
}
private static Risk risk(Map<String,Object> row,String name,Total physical,boolean owner,boolean enabled) {
if(!enabled)return new Risk(null,null,0,null);
int unknown=owner?count(row,name+"_unknown"):row.get(name)==null?1:0;
BigDecimal matched=owner?decimal(row,name+"_count"):BigDecimal.valueOf(count(row,name));
BigDecimal quantity=owner?decimal(row,name+"_physical"):BigDecimal.ONE.equals(matched)?physical.value():BigDecimal.ZERO;
BigDecimal ratio=unknown==0&&physical.value()!=null&&physical.value().signum()>0&&quantity!=null
?quantity.divide(physical.value(),6,RoundingMode.HALF_UP):null;
return new Risk(matched,quantity,unknown,ratio);
}
static Row row(Map<String,Object> raw,InventoryTurnoverRequest q,Context context) {
var r=normalized(raw);boolean owner="OWNER".equals(q.group());
Total physical=total(r,"physical",owner),available=total(r,"available",owner);
var windows=windows(r,context.endDate().plusDays(1));var base=windows.stream().filter(w->w.days()==q.days()).findFirst().orElseThrow();
var coverage=owner?null:scale(decimal(r,"coverage_days"));var average=scale(decimal(r,"average_physical"));
var turnover=scale(decimal(r,"turnover"));int sampleCount=count(r,"sample_count");
String coverageStatus=owner?"NOT_APPLICABLE":available.value()==null?"INVENTORY_UNKNOWN_OR_STALE":available.value().signum()==0?"NO_AVAILABLE_STOCK":
base.outboundQuantity()==null||count(r,"unassessable")>0?"UNASSESSABLE":base.outboundQuantity().signum()==0?"NO_OUTBOUND_IN_WINDOW":"VALID";
String sampleStatus=sampleCount<q.days()?"INSUFFICIENT":average!=null&&average.signum()==0?"ZERO_AVERAGE_STOCK":turnover==null?"SALES_WINDOW_UNAVAILABLE":"VALID";
List<LocalDate> missing=new ArrayList<>();String dates=text(r,"missing_dates");
if(dates!=null&&!dates.isBlank())for(String date:dates.split(","))missing.add(LocalDate.parse(date.substring(0,10)));
List<String> attention=new ArrayList<>();
if(count(r,owner?"no_outbound_count":"no_outbound")>0)attention.add("NO_OUTBOUND_30D");
if(count(r,owner?"slow_count":"slow")>0)attention.add("COVERAGE_GT_THRESHOLD");
if(decimal(r,"frozen")!=null&&decimal(r,"frozen").signum()>0)attention.add("FROZEN_STOCK");
if(decimal(r,"defective")!=null&&decimal(r,"defective").signum()>0)attention.add("DEFECTIVE_STOCK");
if(count(r,"unassessable")>0)attention.add("UNASSESSABLE");
String last=text(r,"last_outbound");
return new Row(text(r,"customer_code"),null,owner?null:text(r,"sku_code"),owner?null:text(r,"product_name"),physical,available,
total(r,"frozen",owner),total(r,"defective",owner),owner?decimal(r,"physical_skus"):physical.value()==null?null:physical.value().signum()>0?BigDecimal.ONE:BigDecimal.ZERO,
windows,base.averageDailyOutboundQuantity(),coverage,coverageStatus,average,turnover,sampleStatus,sampleCount,q.days(),List.copyOf(missing),
"VALID".equals(sampleStatus)?List.of():List.of(sampleStatus),last==null?null:LocalDate.parse(last.substring(0,10)),List.copyOf(attention),
risk(r,"no_outbound",physical,owner,true),risk(r,"slow",physical,owner,q.coverageDaysThreshold()!=null),count(r,"unassessable"));
}
static Overview overview(Map<String,Object> raw) {
var r=normalized(raw);int noUnknown=count(r,"no_outbound_unknown");var noKnown=decimal(r,"no_outbound_count");
return new Overview(total(r,"physical",true),total(r,"available",true),
new Total(noUnknown==0?noKnown:null,noKnown,noUnknown,noUnknown==0?List.of():List.of("SALES_OR_STOCK_UNASSESSABLE")),
new Total(decimal(r,"unassessable"),decimal(r,"unassessable"),0,List.of()));
}
}
@@ -0,0 +1,33 @@
package com.baiyue.lastleg.selfoperated.analytics.query;
import java.util.*;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Isolation;
import org.springframework.transaction.annotation.Transactional;
import static com.baiyue.lastleg.selfoperated.analytics.query.InventoryTurnoverResponses.*;
import static com.baiyue.lastleg.selfoperated.analytics.query.InventoryTurnoverPolicy.*;
@Service
@Transactional(readOnly=true,isolation=Isolation.REPEATABLE_READ)
public class InventoryTurnoverQueryService {
private final InventoryTurnoverContext contexts;
private final InventoryTurnoverReadMapper mapper;
public InventoryTurnoverQueryService(InventoryTurnoverContext contexts,InventoryTurnoverReadMapper mapper){this.contexts=contexts;this.mapper=mapper;}
public Response query(InventoryTurnoverRequest request) {
var frozen=contexts.resolve(request);var input=InventoryTurnoverSql.input(request,frozen);var context=frozen.response();
long total=mapper.count(input);
var rows=request.offset()>=total?List.<Row>of():mapper.page(input).stream().map(r->row(r,request,context)).toList();
Overview summary=null;Charts charts=null;
if(request.overview()) {
summary=overview(mapper.overview(input));
List<OwnerPhysical> owners=new ArrayList<>();int unknown=0;
for(var raw:mapper.owners(input)) {
var r=normalized(raw);
if("META".equals(text(r,"kind")))unknown=count(r,"unknown_count");
else owners.add(new OwnerPhysical(text(r,"customer_code"),null,decimal(r,"physical")));
}
charts=new Charts(List.copyOf(owners),unknown,windows(normalized(mapper.chartWindows(input)),context.endDate().plusDays(1)));
}
return new Response(context,summary,charts,rows,total);
}
}
@@ -0,0 +1,45 @@
package com.baiyue.lastleg.selfoperated.analytics.query;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import org.apache.ibatis.annotations.*;
@Mapper
public interface InventoryTurnoverReadMapper {
record SampleRun(LocalDate sampleDate,Long runId) {}
@Select("SELECT "+AnalyticsReadMapper.RUN_COLUMNS+" FROM ll_selfop_sync_run WHERE scope_id=#{scopeId} "
+"AND scope_hash=#{scopeHash} AND status='PUBLISHED' AND run_role='WAREHOUSE_SYNC' AND execution_purpose='BUSINESS' "
+"AND published_at_utc<=#{asOf} ORDER BY published_at_utc DESC,id DESC LIMIT 1")
AnalyticsReadRows.Run versionAt(@Param("scopeId") Long scopeId,@Param("scopeHash") String scopeHash,@Param("asOf") LocalDateTime asOf);
// The immutable first-daily marker is set only by publication. Recheck its original schedule evidence.
String SAMPLE_WHERE=" r.source_account=#{account} AND r.warehouse_code=#{warehouse} "
+"AND r.inventory_sample_rule_version='DAILY_SCHEDULED_FIRST_V1' AND r.status='PUBLISHED' "
+"AND r.execution_purpose='BUSINESS' AND r.run_role='WAREHOUSE_SYNC' AND r.coverage_status='COMPLETE' "
+"AND r.published_at_utc<=#{asOf} AND t.execution_purpose='BUSINESS' AND t.task_type='SCHEDULED' "
+"AND t.task_role='WAREHOUSE_SYNC' AND r.page_count>=1 AND r.source_total_count>=0 "
+"AND r.source_total_count=r.unique_object_count AND r.fetch_started_at_utc<=r.fetch_finished_at_utc "
+"AND DATE(DATE_ADD(t.planned_at_utc,INTERVAL 8 HOUR))=r.inventory_daily_sample_date "
+"AND DATE(DATE_ADD(r.fetch_started_at_utc,INTERVAL 8 HOUR))=r.inventory_daily_sample_date "
+"AND DATE(DATE_ADD(r.fetch_finished_at_utc,INTERVAL 8 HOUR))=r.inventory_daily_sample_date ";
@Select("SELECT r.inventory_daily_sample_date sample_date,r.id run_id FROM ll_selfop_sync_run r "
+"JOIN ll_selfop_sync_task t ON t.id=r.task_id WHERE "+SAMPLE_WHERE
+"AND r.inventory_daily_sample_date>=#{from} AND r.inventory_daily_sample_date<#{until} ORDER BY r.inventory_daily_sample_date")
List<SampleRun> samples(@Param("account") String account,@Param("warehouse") String warehouse,
@Param("asOf") LocalDateTime asOf,@Param("from") LocalDate from,@Param("until") LocalDate until);
@Select("SELECT MIN(r.inventory_daily_sample_date) FROM ll_selfop_sync_run r JOIN ll_selfop_sync_task t ON t.id=r.task_id WHERE "+SAMPLE_WHERE)
LocalDate accumulationStartedOn(@Param("account") String account,@Param("warehouse") String warehouse,@Param("asOf") LocalDateTime asOf);
@SelectProvider(type=InventoryTurnoverSql.class,method="page")
List<Map<String,Object>> page(@Param("p") InventoryTurnoverSql.Input input);
@SelectProvider(type=InventoryTurnoverSql.class,method="count")
long count(@Param("p") InventoryTurnoverSql.Input input);
@SelectProvider(type=InventoryTurnoverSql.class,method="overview")
Map<String,Object> overview(@Param("p") InventoryTurnoverSql.Input input);
@SelectProvider(type=InventoryTurnoverSql.class,method="owners")
List<Map<String,Object>> owners(@Param("p") InventoryTurnoverSql.Input input);
@SelectProvider(type=InventoryTurnoverSql.class,method="chartWindows")
Map<String,Object> chartWindows(@Param("p") InventoryTurnoverSql.Input input);
}
@@ -0,0 +1,49 @@
package com.baiyue.lastleg.selfoperated.analytics.query;
import com.baiyue.common.core.exception.BizException;
import java.time.LocalDate;
import java.util.Set;
import org.springframework.format.annotation.DateTimeFormat;
/** Separate from DETAIL: its dates, windows and filters must never be silently mixed. */
public record InventoryTurnoverRequest(String sourceAccount, String warehouseCode,
@DateTimeFormat(iso=DateTimeFormat.ISO.DATE) LocalDate endDate, Integer baseWindowDays,
String groupBy, String customerCode, String sku, String focus, Integer coverageDaysThreshold,
String turnoverStatus, String sortBy, Integer current, Integer size, Boolean includeOverview,
String snapshotContext) {
private static final Set<String> PARAMETERS=Set.of("sourceAccount","warehouseCode","endDate","baseWindowDays",
"groupBy","customerCode","sku","focus","coverageDaysThreshold","turnoverStatus","sortBy","current","size","includeOverview","snapshotContext");
public static void validateParameters(Set<String> keys) {
if(!PARAMETERS.containsAll(keys)) throw new BizException("库存周转不接受其他视图的查询参数");
}
public int days(){return baseWindowDays==null?30:baseWindowDays;}
public String group(){return groupBy==null?"SKU":groupBy;}
public String attention(){return focus==null?"ALL":focus;}
public String sort(){return sortBy==null?"PHYSICAL_DESC":sortBy;}
public int page(){return current==null?1:current;}
public int pageSize(){return size==null?20:size;}
public long offset(){return (long)(page()-1)*pageSize();}
public boolean overview(){return snapshotContext==null||!Boolean.FALSE.equals(includeOverview);}
public LocalDate end(LocalDate today){return endDate==null?today.minusDays(1):endDate;}
public void validate(LocalDate today) {
AnalyticsQuery.text(sourceAccount,64,"来源账号"); AnalyticsQuery.text(warehouseCode,64,"仓库");
if(customerCode!=null) AnalyticsQuery.text(customerCode,64,"货主");
if(sku!=null) AnalyticsQuery.text(sku,128,"SKU");
if(!Set.of(7,15,30).contains(days())) throw new BizException("基准窗口仅支持7、15、30日");
if(!end(today).isBefore(today)) throw new BizException("截止日必须是北京时间已结束的日期");
if(!Set.of("SKU","OWNER").contains(group())) throw new BizException("分组仅支持SKU或OWNER");
if("OWNER".equals(group())&&sku!=null) throw new BizException("货主汇总不接受SKU筛选");
if(!Set.of("ALL","NO_OUTBOUND_30D","COVERAGE_GT","COVERAGE_LE","FROZEN","DEFECTIVE","UNASSESSABLE").contains(attention()))
throw new BizException("库存关注条件无效");
if(coverageDaysThreshold!=null&&coverageDaysThreshold<=0 || attention().startsWith("COVERAGE_")&&coverageDaysThreshold==null)
throw new BizException("可售天数阈值须为正整数");
if(turnoverStatus!=null&&!"INSUFFICIENT".equals(turnoverStatus)) throw new BizException("周转状态无效");
if(!Set.of("PHYSICAL_DESC","AVAILABLE_DESC","OUTBOUND_7_DESC","OUTBOUND_15_DESC","OUTBOUND_30_DESC",
"COVERAGE_ASC","COVERAGE_DESC","TURNOVER_DESC","NO_OUTBOUND_PHYSICAL_DESC").contains(sort())
|| "OWNER".equals(group())&&sort().startsWith("COVERAGE_")
|| "SKU".equals(group())&&"NO_OUTBOUND_PHYSICAL_DESC".equals(sort())) throw new BizException("排序不适用于当前分组");
if(page()<1||pageSize()<1||pageSize()>100) throw new BizException("分页范围为current≥1,size为1至100");
if(snapshotContext==null&&page()!=1) throw new BizException("翻页必须携带首屏上下文");
if(snapshotContext!=null&&(snapshotContext.isBlank()||snapshotContext.length()>16384)) throw new BizException("查询上下文无效,请整体刷新");
}
}
@@ -0,0 +1,31 @@
package com.baiyue.lastleg.selfoperated.analytics.query;
import java.math.BigDecimal;
import java.time.Instant;
import java.time.LocalDate;
import java.util.List;
public final class InventoryTurnoverResponses {
private InventoryTurnoverResponses() {}
public record Window(int days, LocalDate startDate, LocalDate endDateExclusive, BigDecimal outboundQuantity,
BigDecimal observedQuantity, BigDecimal averageDailyOutboundQuantity, int completeDayCount,
int partialDayCount, int unknownDayCount, String dataStatus, List<String> reasonCodes) {}
public record Total(BigDecimal value, BigDecimal knownSubtotal, int unknownSkuCount, List<String> reasonCodes) {}
public record Risk(BigDecimal matchedSkuCount, BigDecimal knownPhysicalQuantity, int unknownSkuCount, BigDecimal physicalQuantityRatio) {}
public record Row(String customerCode,String customerName,String sku,String productName,
Total physicalQuantity,Total availableQuantity,Total frozenQuantity,Total defectiveQuantity,
BigDecimal physicalSkuCount,List<Window> windows,BigDecimal averageDailyOutboundQuantity,
BigDecimal inventoryCoverageDays,String coverageStatus,BigDecimal averagePhysicalQuantity,
BigDecimal quantityTurnoverCount,String turnoverStatus,int qualifiedSampleDays,int requiredSampleDays,
List<LocalDate> missingSampleDates,List<String> sampleReasonCodes,LocalDate lastObservedOutboundDate,
List<String> attentionReasons,Risk noOutbound30d,Risk slowCoverage,int unassessableSkuCount) {}
public record Sample(LocalDate date,Long runId) {}
public record Context(String snapshotContext,String sourceAccount,String warehouseCode,LocalDate endDate,
int baseWindowDays,Instant queryAsOf,Instant queryExpiresAt,AnalyticsResponses.Context analyticsContext,
Instant inventoryObservedFrom,Instant inventoryObservedUntil,String sampleRuleVersion,
List<Sample> dailySamples,List<LocalDate> missingSampleDates,LocalDate accumulationStartedOn) {}
public record Overview(Total physicalQuantity,Total availableQuantity,Total noOutboundSkuCount,Total unassessableSkuCount) {}
public record OwnerPhysical(String customerCode,String customerName,BigDecimal physicalQuantity) {}
public record Charts(List<OwnerPhysical> ownerPhysicalTop,int unknownOwnerCount,List<Window> averageDaily) {}
public record Response(Context context,Overview summary,Charts charts,List<Row> rows,long total) {}
}
@@ -0,0 +1,158 @@
package com.baiyue.lastleg.selfoperated.analytics.query;
import java.time.*;
import java.util.*;
/** Every value is bound. Only validated enums select SQL fragments; no source order/remote API reads. */
public final class InventoryTurnoverSql {
private InventoryTurnoverSql() {}
public record Day(LocalDate date,LocalDateTime startUtc,int coverage,Long sampleRun) {}
public record Input(InventoryTurnoverRequest query,long salesRun,long inventoryRun,boolean stockFresh,
boolean salesKnown,List<Day> days) {
public LocalDate until(){return days.get(29).date().plusDays(1);}
public LocalDate from(){return days.get(0).date();}
public LocalDate baseFrom(){return until().minusDays(query.days());}
public LocalDate from7(){return until().minusDays(7);}
public LocalDate from15(){return until().minusDays(15);}
public LocalDate from30(){return from();}
public LocalDateTime baseFromUtc(){return baseFrom().atStartOfDay().minusHours(8);}
public LocalDateTime fromUtc(){return from().atStartOfDay().minusHours(8);}
}
static Input input(InventoryTurnoverRequest q,InventoryTurnoverContext.Frozen frozen) {
var coverage=SalesQueryCoverage.intervals(frozen.snapshot().sales());
boolean trusted=frozen.snapshot().sourceKnown()&&!frozen.response().analyticsContext().rebuildRequired();
var days=frozen.response().dailySamples().stream().map(s->new Day(s.date(),s.date().atStartOfDay().minusHours(8),
!trusted?0:SalesQueryCoverage.complete(coverage,s.date(),s.date().plusDays(1))?1:SalesQueryCoverage.at(coverage,s.date())==null?0:2,s.runId())).toList();
return new Input(q,frozen.response().analyticsContext().salesRunId(),frozen.response().analyticsContext().inventoryRunId(),
"VALID".equals(frozen.response().analyticsContext().inventoryStatus()),frozen.snapshot().sourceKnown(),days);
}
private static String scope(String alias) {
return alias+"source_account=#{p.query.sourceAccount} AND "+alias+"warehouse_code=#{p.query.warehouseCode}"
+" AND (#{p.query.customerCode,jdbcType=VARCHAR} IS NULL OR "+alias+"customer_code=#{p.query.customerCode,jdbcType=VARCHAR})";
}
private static String valid(String field){return field+" IS NOT NULL AND "+field+">=0 AND (quality_status='VALID' OR quality_reason IN ('FIELD_MISSING','NEGATIVE_QUANTITY'))";}
private static String amount(String field){return "CASE WHEN MIN(CASE WHEN "+valid(field)+" THEN 1 ELSE 0 END)=1 THEN SUM("+field+") END";}
private static String calendar(Input p) {
var parts=new ArrayList<String>();
for(int i=0;i<p.days().size();i++) parts.add("SELECT CAST(#{p.days["+i+"].date} AS DATE) business_date,CAST(#{p.days["+i+"].startUtc} AS DATETIME) start_utc,"
+"CAST(#{p.days["+i+"].coverage} AS SIGNED) coverage,CAST(#{p.days["+i+"].sampleRun,jdbcType=BIGINT} AS SIGNED) sample_run");
return String.join(" UNION ALL ",parts);
}
private static String cte(Input p) {
String stocks="SELECT snapshot_run_id,customer_code,sku_code,"+amount("physical_quantity")+" physical,"
+amount("frozen_quantity")+" frozen,MAX(CASE WHEN product_quality=0 AND "+valid("available_quantity")+" THEN available_quantity END) available,"
+"CASE WHEN MIN(CASE WHEN product_quality NOT IN (0,1) OR product_quality IS NULL THEN 0 WHEN product_quality=1 AND NOT ("+valid("physical_quantity")+") THEN 0 ELSE 1 END)=1 "
+"THEN SUM(CASE WHEN product_quality=1 THEN physical_quantity ELSE 0 END) END defective "
+"FROM ll_selfop_inventory_snapshot WHERE "+scope("")+" AND (snapshot_run_id=#{p.inventoryRun} OR snapshot_run_id IN (SELECT sample_run FROM calendar)) "
+"GROUP BY snapshot_run_id,customer_code,sku_code";
String profiles="SELECT * FROM ll_selfop_sku_profile WHERE "+scope("")+" AND (profile_run_id IN (#{p.salesRun},#{p.inventoryRun}) OR profile_run_id IN (SELECT sample_run FROM calendar))";
StringBuilder sql=new StringBuilder("WITH calendar AS (").append(calendar(p)).append("), fixed_profiles AS (").append(profiles)
.append("), stocks AS (").append(stocks).append("), sales_days AS (SELECT customer_code,sku_code,business_date,SUM(outbound_quantity) quantity,MIN(CASE WHEN quantity_unit='PCS' AND outbound_quantity>=0 THEN 1 ELSE 0 END) unit_valid FROM ll_selfop_sku_daily_sales WHERE ")
.append(scope("")).append(" AND sales_run_id=#{p.salesRun} AND business_date>=CAST(#{p.from} AS DATE) AND business_date<CAST(#{p.until} AS DATE) GROUP BY customer_code,sku_code,business_date), candidates AS (")
.append("SELECT customer_code,sku_code FROM fixed_profiles UNION SELECT customer_code,sku_code FROM stocks UNION SELECT customer_code,sku_code FROM sales_days), profiles AS (")
.append("SELECT c.customer_code,c.sku_code,MAX(f.product_name) product_name,MIN(CASE WHEN f.quantity_unit='PCS' THEN 1 ELSE 0 END) unit_valid,")
.append("MIN(CASE WHEN f.evidence_run_id IS NOT NULL AND f.evidence_type IN ('VALID_OUTBOUND','INVENTORY_OBSERVATION') THEN f.earliest_evidence_at_utc END) evidence ")
.append("FROM candidates c LEFT JOIN fixed_profiles f ON f.customer_code=c.customer_code AND f.sku_code=c.sku_code GROUP BY c.customer_code,c.sku_code), daily_values AS (")
.append("SELECT c.customer_code,c.sku_code,d.business_date,d.coverage,c.unit_valid,c.evidence,sd.quantity,sd.unit_valid sales_unit,")
.append("CASE WHEN ip.quantity_unit='PCS' THEN hs.physical END sample_physical ")
.append("FROM profiles c CROSS JOIN calendar d LEFT JOIN sales_days sd ON sd.customer_code=c.customer_code AND sd.sku_code=c.sku_code AND sd.business_date=d.business_date ")
.append("LEFT JOIN stocks hs ON hs.snapshot_run_id=d.sample_run AND hs.customer_code=c.customer_code AND hs.sku_code=c.sku_code ")
.append("LEFT JOIN fixed_profiles ip ON ip.profile_run_id=d.sample_run AND ip.customer_code=c.customer_code AND ip.sku_code=c.sku_code), day_result AS (")
.append("SELECT v.*,CASE WHEN #{p.salesKnown}=1 AND unit_valid=1 AND COALESCE(sales_unit,1)=1 THEN CASE WHEN quantity IS NOT NULL THEN quantity WHEN coverage>0 AND evidence<=CONVERT_TZ(CAST(business_date AS DATETIME),'+08:00','+00:00') THEN 0 END END observed,")
.append("CASE WHEN coverage=1 AND unit_valid=1 AND COALESCE(sales_unit,1)=1 AND (quantity IS NOT NULL OR evidence<=CONVERT_TZ(CAST(business_date AS DATETIME),'+08:00','+00:00')) THEN 1 ELSE 0 END complete,")
.append("CASE WHEN coverage=1 AND unit_valid=1 AND COALESCE(sales_unit,1)=1 THEN 1 ELSE 0 END owner_complete FROM daily_values v), sku_windows AS (")
.append("SELECT customer_code,sku_code").append(windowAggregates(false)).append(",COUNT(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END) sample_count,")
.append("CASE WHEN COUNT(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END)=#{p.query.days} THEN SUM(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END)/#{p.query.days} END average_physical,")
.append("GROUP_CONCAT(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) AND sample_physical IS NULL THEN CAST(business_date AS CHAR) END ORDER BY business_date SEPARATOR ',') missing_dates ")
.append("FROM day_result GROUP BY customer_code,sku_code), last_sales AS (SELECT customer_code,sku_code,MAX(business_date) last_outbound FROM ll_selfop_sku_daily_sales WHERE ")
.append(scope("")).append(" AND sales_run_id=#{p.salesRun} AND business_date<CAST(#{p.until} AS DATE) AND outbound_quantity>0 AND quantity_unit='PCS' GROUP BY customer_code,sku_code)");
sql.append(", sku_base AS (SELECT c.customer_code,c.sku_code,c.product_name,c.evidence,c.unit_valid");
for(int n:List.of(7,15,30)) for(String suffix:List.of("outbound","observed","complete","partial")) sql.append(",w.w").append(n).append('_').append(suffix);
sql.append(",w.sample_count,w.average_physical,w.missing_dates,ls.last_outbound");
for(String field:List.of("physical","available","frozen","defective")) sql.append(",CASE WHEN #{p.stockFresh}=1 AND ip.quantity_unit='PCS' THEN st.").append(field).append(" END ").append(field);
sql.append(" FROM profiles c JOIN sku_windows w ON w.customer_code=c.customer_code AND w.sku_code=c.sku_code LEFT JOIN last_sales ls ON ls.customer_code=c.customer_code AND ls.sku_code=c.sku_code LEFT JOIN stocks st ON st.snapshot_run_id=#{p.inventoryRun} AND st.customer_code=c.customer_code AND st.sku_code=c.sku_code ")
.append("LEFT JOIN fixed_profiles ip ON ip.profile_run_id=#{p.inventoryRun} AND ip.customer_code=c.customer_code AND ip.sku_code=c.sku_code), sku_calculated AS (SELECT b.*,")
.append("CASE WHEN available=0 THEN 0 WHEN available>0 AND w").append(p.query.days()).append("_outbound>0 AND evidence<=#{p.baseFromUtc} THEN available*#{p.query.days}/w").append(p.query.days()).append("_outbound END coverage_days,")
.append("CASE WHEN average_physical>0 AND evidence<=#{p.baseFromUtc} THEN w").append(p.query.days()).append("_outbound/average_physical END turnover,")
.append("CASE WHEN physical=0 THEN 0 WHEN physical>0 AND w30_outbound IS NOT NULL AND evidence<=#{p.fromUtc} THEN CASE WHEN w30_outbound=0 THEN 1 ELSE 0 END END no_outbound,")
.append("CASE WHEN physical IS NULL OR available IS NULL OR w").append(p.query.days()).append("_outbound IS NULL OR evidence IS NULL OR evidence>#{p.baseFromUtc} THEN 1 ELSE 0 END unassessable FROM sku_base b), sku_result AS (")
.append("SELECT b.*,CASE WHEN #{p.query.coverageDaysThreshold,jdbcType=INTEGER} IS NULL THEN NULL WHEN physical=0 OR available=0 THEN 0 WHEN coverage_days IS NOT NULL AND available>0 THEN CASE WHEN coverage_days>#{p.query.coverageDaysThreshold,jdbcType=INTEGER} THEN 1 ELSE 0 END WHEN available>0 AND w")
.append(p.query.days()).append("_outbound=0 AND unassessable=0 THEN 0 END slow FROM sku_calculated b) ");
return sql.toString();
}
private static String windowAggregates(boolean owner) {
StringBuilder sql=new StringBuilder();
for(int n:List.of(7,15,30)) {
String range="business_date>=CAST(#{p.from"+n+"} AS DATE)";
sql.append(",CASE WHEN SUM(CASE WHEN ").append(range).append(" THEN complete ELSE 0 END)=").append(n)
.append(" THEN SUM(CASE WHEN ").append(range).append(" THEN COALESCE(observed,0) END) END w").append(n).append("_outbound")
.append(",SUM(CASE WHEN ").append(range).append(" THEN observed END) w").append(n).append("_observed")
.append(",SUM(CASE WHEN ").append(range).append(" THEN complete ELSE 0 END) w").append(n).append("_complete")
.append(",SUM(CASE WHEN ").append(range).append(" AND complete=0 AND observed IS NOT NULL THEN 1 ELSE 0 END) w").append(n).append("_partial");
}
return sql.toString();
}
private static String ownerCte(Input p) {
return ", owner_days AS (SELECT customer_code,business_date,MIN(owner_complete) complete,"
+"CASE WHEN MIN(owner_complete)=1 THEN SUM(COALESCE(quantity,0)) ELSE SUM(observed) END observed,"
+"CASE WHEN COUNT(sample_physical)=COUNT(*) THEN SUM(sample_physical) END sample_physical FROM day_result GROUP BY customer_code,business_date), owner_windows AS (SELECT customer_code"
+windowAggregates(true)+",COUNT(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END) sample_count,"
+"CASE WHEN COUNT(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END)=#{p.query.days} THEN SUM(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END)/#{p.query.days} END average_physical,"
+"GROUP_CONCAT(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) AND sample_physical IS NULL THEN CAST(business_date AS CHAR) END ORDER BY business_date SEPARATOR ',') missing_dates FROM owner_days GROUP BY customer_code), owner_totals AS ("
+totals("customer_code",true)+" FROM sku_result GROUP BY customer_code), owner_result AS (SELECT t.*,w.sample_count,w.average_physical,w.missing_dates,"
+"CASE WHEN w.average_physical>0 THEN w.w"+p.query.days()+"_outbound/w.average_physical END turnover,"
+String.join(",",List.of(7,15,30).stream().flatMap(n->List.of("outbound","observed","complete","partial").stream().map(s->"w.w"+n+"_"+s)).toList())
+" FROM owner_totals t JOIN owner_windows w ON w.customer_code=t.customer_code) ";
}
private static String totals(String key,boolean withFocus) {
var sql=new StringBuilder("SELECT ").append(key==null?"COUNT(*) sku_count":key);
for(String field:List.of("physical","available","frozen","defective")) sql.append(",CASE WHEN COUNT(").append(field).append(")=COUNT(*) THEN COALESCE(SUM(").append(field).append("),0) END ").append(field)
.append(",SUM(").append(field).append(") ").append(field).append("_known,COUNT(*)-COUNT(").append(field).append(") ").append(field).append("_unknown");
sql.append(",MAX(last_outbound) last_outbound,CASE WHEN COUNT(physical)=COUNT(*) THEN SUM(CASE WHEN physical>0 THEN 1 ELSE 0 END) END physical_skus,COALESCE(SUM(unassessable),0) unassessable");
for(String risk:List.of("no_outbound","slow")) sql.append(",SUM(CASE WHEN ").append(risk).append("=1 THEN 1 ELSE 0 END) ").append(risk).append("_count,SUM(CASE WHEN ").append(risk).append("=1 THEN physical ELSE 0 END) ").append(risk).append("_physical,COUNT(*)-COUNT(").append(risk).append(") ").append(risk).append("_unknown");
if(withFocus)sql.append(",MAX(CASE WHEN available>0 AND coverage_days IS NOT NULL AND coverage_days<=#{p.query.coverageDaysThreshold,jdbcType=INTEGER} THEN 1 ELSE 0 END) fast_count,MAX(CASE WHEN frozen>0 THEN 1 ELSE 0 END) frozen_count,MAX(CASE WHEN defective>0 THEN 1 ELSE 0 END) defective_count");
return sql.toString();
}
private static String filter(Input p,boolean owner) {
String predicate=switch(p.query.attention()) {
case "NO_OUTBOUND_30D" -> owner?"no_outbound_count>0":"no_outbound=1";
case "COVERAGE_GT" -> owner?"slow_count>0":"slow=1";
case "COVERAGE_LE" -> owner?"fast_count>0":"available>0 AND coverage_days IS NOT NULL AND coverage_days<=#{p.query.coverageDaysThreshold}";
case "FROZEN" -> owner?"frozen_count>0":"frozen>0";
case "DEFECTIVE" -> owner?"defective_count>0":"defective>0";
case "UNASSESSABLE" -> "unassessable>0";
default -> "1=1";
};
if(p.query.turnoverStatus()!=null) predicate+=" AND sample_count<#{p.query.days}";
if(!owner&&p.query.sku()!=null) predicate+=" AND sku_code=#{p.query.sku}";
return " WHERE "+predicate;
}
private static String order(Input p,boolean owner) {
String field=switch(p.query.sort()) {
case "AVAILABLE_DESC" -> "available";case "OUTBOUND_7_DESC" -> "w7_outbound";
case "OUTBOUND_15_DESC" -> "w15_outbound";case "OUTBOUND_30_DESC" -> "w30_outbound";
case "COVERAGE_ASC","COVERAGE_DESC" -> "coverage_days";case "TURNOVER_DESC" -> "turnover";
case "NO_OUTBOUND_PHYSICAL_DESC" -> "no_outbound_physical";default -> "physical";
};
return " ORDER BY "+field+" IS NULL,"+field+("COVERAGE_ASC".equals(p.query.sort())?" ASC":" DESC")+(owner?",customer_code":",sku_code,customer_code");
}
public static String page(Map<String,Object> args) {
Input p=(Input)args.get("p");boolean owner="OWNER".equals(p.query.group());
return cte(p)+(owner?ownerCte(p):"")+"SELECT * FROM "+(owner?"owner_result":"sku_result")+filter(p,owner)+order(p,owner)+" LIMIT #{p.query.pageSize} OFFSET #{p.query.offset}";
}
public static String count(Map<String,Object> args) {
Input p=(Input)args.get("p");boolean owner="OWNER".equals(p.query.group());
return cte(p)+(owner?ownerCte(p):"")+"SELECT COUNT(*) FROM "+(owner?"owner_result":"sku_result")+filter(p,owner);
}
public static String overview(Map<String,Object> args) {
Input p=(Input)args.get("p");return cte(p)+totals(null,false)+" FROM sku_result";
}
public static String owners(Map<String,Object> args) {
Input p=(Input)args.get("p");return cte(p)+ownerCte(p)
+", ranked_owners AS (SELECT customer_code,physical FROM owner_result WHERE physical IS NOT NULL ORDER BY physical DESC,customer_code LIMIT 10) "
+"SELECT 'OWNER' kind,customer_code,physical,0 unknown_count FROM ranked_owners UNION ALL SELECT 'META' kind,NULL customer_code,NULL physical,COUNT(*) unknown_count FROM owner_result WHERE physical IS NULL";
}
public static String chartWindows(Map<String,Object> args) {
Input p=(Input)args.get("p");return cte(p)+", warehouse_days AS (SELECT business_date,MIN(owner_complete) complete,CASE WHEN MIN(owner_complete)=1 THEN SUM(COALESCE(quantity,0)) ELSE SUM(observed) END observed FROM day_result GROUP BY business_date) SELECT COUNT(*) day_count"+windowAggregates(true)+" FROM warehouse_days";
}
}