Skip to content

D
dc-java
Commit 2c816ee7 by huluobin
Options

亚马逊广告生成 

亚马逊广告下载
帐号同步
parent 1ca04f19
Showing with 259 additions and 121 deletions
data-base/base-sync-amazon-ad/src/main/resources/job.properties
#EVENT_RDB_STORAGE_DRIVER=com.mysql.jdbc.Driver
#EVENT_RDB_STORAGE_URL=jdbc:mysql://127.0.0.1:3306/bailun_datacenter?useUnicode=true&characterEncoding=UTF-8&serverTimeZone=GMT&useSSL=false&allowPublicKeyRetrieval=true&allowMultiQueries=true
#EVENT_RDB_STORAGE_URL=jdbc:mysql://gz-cdb-kp7s5i79.sql.tencentcdb.com:61691/bailun_datacenter?useUnicode=true&characterEncoding=utf-8
#EVENT_RDB_STORAGE_USERNAME=root
#EVENT_RDB_STORAGE_PASSWORD=root
#ZOOKEEPER_SERVER=94.191.111.219:2181
#db
EVENT_RDB_STORAGE_DRIVER=com.mysql.jdbc.Driver
EVENT_RDB_STORAGE_URL=jdbc:mysql://10.0.8.15:3306/bailun_datacenter?useUnicode=true&characterEncoding=utf-8&serverTimezone=GMT%2B8
EVENT_RDB_STORAGE_USERNAME=root
EVENT_RDB_STORAGE_PASSWORD=#7kfnymAM$Y9-Ntf
#zk
ZOOKEEPER_SERVER=172.31.255.120:2181
NAME_SPACE=data-center
JOB_NAME_AD_PRODUCT=base-sync-finance-amazon-ad-product
JOB_CRON_AD_PRODUCT=0 1 0/5 * * ? *
#SHARDING_TOTAL_COUNT=3
# 每天13点生成昨天的报告
JOB_NAME_AD_PRODUCT_GENERATE=base-amazon-ad-product-generate
JOB_CRON_AD_PRODUCT_GENERATE=0 1 12/1 * * ? *
JOB_CRON_AD_PRODUCT_GENERATE=0 0 13 * * ?
#每小时下载报告
JOB_NAME_AD_PRODUCT_DOWNLOAD=base-amazon-ad-product-download
JOB_CRON_AD_PRODUCT_DOWNLOAD=0 1 0/1 * * ? *
#任务分片总数
SHARDING_TOTAL_COUNT=2
data-base/base-sync-amazon-ad/src/test/java/AmazonAdProductTest.java
......@@ -112,15 +112,9 @@ public class AmazonAdProductTest {
AmazonAdDownloadReportJob amazonAdDownloadReportJob = new AmazonAdDownloadReportJob();
amazonAdGenerateReportIdJob.executeJob(new ShardingContext(
amazonAdGenerateReportIdJob.execute(new ShardingContext(
new ShardingContexts(null, "amazonAdGenerateReportIdJob", 1, null, new HashMap<>()), 0
), new JobPointLog("amazonAdGenerateReportIdJob",
1,
100,
null,
1,
LocalDateTime.of(2020, 4, 1, 0, 0),
LocalDateTime.of(2020, 4, 2, 0, 0)));
));
amazonAdDownloadReportJob.execute(new ShardingContext(
......
data-base/base-sync-amazon-ad/src/test/java/com/bailuntec/job/AmazonAdDownloadReportJobTest.java
......@@ -53,4 +53,5 @@ class AmazonAdDownloadReportJobTest {
List<AmazonAdProduct> amazonAdProductList = new Gson().fromJson(new InputStreamReader(inputStream), type);
log.warn("的Report解析完成");
}
}
data-base/base-sync-amazon-ad/src/test/java/com/bailuntec/job/AmazonAdGenerateReportIdJobTest.java 0 → 100644
package com.bailuntec.job;
import com.bailuntec.domain.entity.JobPointLog;
import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.executor.ShardingContexts;
import org.junit.jupiter.api.Test;
import java.time.LocalDateTime;
import java.util.HashMap;
/**
* <p>
*
* </p>
*
* @author robbendev
* @since 2020/8/24 9:31 上午
*/
public class AmazonAdGenerateReportIdJobTest {
@Test
public void executeJob() {
AmazonAdGenerateReportIdJob amazonAdGenerateReportIdJob = new AmazonAdGenerateReportIdJob();
amazonAdGenerateReportIdJob.execute(new ShardingContext(new ShardingContexts("x", null, 1, null, new HashMap<>()), 0));
}
}
data-base/base-sync-company/src/main/java/com/bailuntec/Application.java
......@@ -26,6 +26,7 @@ public class Application {
private static final String EVENT_RDB_STORAGE_URL = propertiesUtil.getPropertyAsString("EVENT_RDB_STORAGE_URL");
private static final String EVENT_RDB_STORAGE_USERNAME = propertiesUtil.getPropertyAsString("EVENT_RDB_STORAGE_USERNAME");
private static final String EVENT_RDB_STORAGE_PASSWORD = propertiesUtil.getPropertyAsString("EVENT_RDB_STORAGE_PASSWORD");
public static void main(String[] args) {
new JobScheduler(createRegistryCenter(), createJobConfiguration1()).init();
new JobScheduler(createRegistryCenter(), createJobConfiguration2()).init();
......@@ -40,23 +41,33 @@ public class Application {
}
private static LiteJobConfiguration createJobConfiguration1() {
JobCoreConfiguration simpleCoreConfig = JobCoreConfiguration.newBuilder(propertiesUtil.getPropertyAsString("JOB_NAME_COMPANY_INFO"), propertiesUtil.getPropertyAsString("JOB_CRON_COMPANY_INFO"), propertiesUtil.getPropertyAsInt("SHARDING_TOTAL_COUNT")).build();
JobCoreConfiguration simpleCoreConfig = JobCoreConfiguration.newBuilder(
propertiesUtil.getPropertyAsString("JOB_NAME_COMPANY_INFO"),
propertiesUtil.getPropertyAsString("JOB_CRON_COMPANY_INFO"),
propertiesUtil.getPropertyAsInt("SHARDING_TOTAL_COUNT"))
.build();
SimpleJobConfiguration simpleJobConfig = new SimpleJobConfiguration(simpleCoreConfig, CompanyInfoSyncJob.class.getCanonicalName());
LiteJobConfiguration simpleJobRootConfig = LiteJobConfiguration.newBuilder(simpleJobConfig).build();
return simpleJobRootConfig;
return LiteJobConfiguration.newBuilder(simpleJobConfig).build();
}
private static LiteJobConfiguration createJobConfiguration2() {
JobCoreConfiguration simpleCoreConfig = JobCoreConfiguration.newBuilder(propertiesUtil.getPropertyAsString("JOB_NAME_COMPANY_STAFF"), propertiesUtil.getPropertyAsString("JOB_CRON_COMPANY_STAFF"), propertiesUtil.getPropertyAsInt("SHARDING_TOTAL_COUNT")).build();
SimpleJobConfiguration simpleJobConfig = new SimpleJobConfiguration(simpleCoreConfig, CompanyStaffSyncJob.class.getCanonicalName());
LiteJobConfiguration simpleJobRootConfig = LiteJobConfiguration.newBuilder(simpleJobConfig).build();
return simpleJobRootConfig;
}
private static LiteJobConfiguration createJobConfiguration3() {
JobCoreConfiguration simpleCoreConfig = JobCoreConfiguration.newBuilder(propertiesUtil.getPropertyAsString("JOB_NAME_COMPANY_ACCOUNT"), propertiesUtil.getPropertyAsString("JOB_CRON_COMPANY_ACCOUNT"), propertiesUtil.getPropertyAsInt("SHARDING_TOTAL_COUNT")).build();
JobCoreConfiguration simpleCoreConfig = JobCoreConfiguration.newBuilder(
propertiesUtil.getPropertyAsString("JOB_NAME_COMPANY_ACCOUNT"),
propertiesUtil.getPropertyAsString("JOB_CRON_COMPANY_ACCOUNT"),
propertiesUtil.getPropertyAsInt("SHARDING_TOTAL_COUNT")).build();
SimpleJobConfiguration simpleJobConfig = new SimpleJobConfiguration(simpleCoreConfig, CompanyAccountSyncJob.class.getCanonicalName());
LiteJobConfiguration simpleJobRootConfig = LiteJobConfiguration.newBuilder(simpleJobConfig).build();
return simpleJobRootConfig;
}
private static LiteJobConfiguration createJobConfiguration4() {
JobCoreConfiguration simpleCoreConfig = JobCoreConfiguration.newBuilder(propertiesUtil.getPropertyAsString("JOB_NAME_PAYPAL_ACCOUNT"), propertiesUtil.getPropertyAsString("JOB_CRON_PAYPAL_ACCOUNT"), propertiesUtil.getPropertyAsInt("SHARDING_TOTAL_COUNT")).build();
SimpleJobConfiguration simpleJobConfig = new SimpleJobConfiguration(simpleCoreConfig, PaypalAccountSyncJob.class.getCanonicalName());
......
data-base/base-sync-company/src/main/java/com/bailuntec/job/CompanyAccountSyncJob.java
package com.bailuntec.job;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.bailuntec.domain.entity.DcBaseCompanyAccount;
import com.bailuntec.domain.entity.JobPointLog;
import com.bailuntec.domain.example.DcBaseCompanyAccountExample;
import com.bailuntec.domain.pojo.AccountResult;
import com.bailuntec.domain.pojo.CompanyAccountInfo;
import com.bailuntec.mapper.DcBaseCompanyAccountMapper;
import com.bailuntec.support.PointJob;
import com.bailuntec.utils.OkHttpUtil;
import com.bailuntec.utils.PropertiesUtil;
import com.bailuntec.utils.SessionUtil;
import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import lombok.extern.slf4j.Slf4j;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.ibatis.session.SqlSession;
import java.io.IOException;
import java.time.LocalDateTime;
import java.util.List;
@Slf4j
public class CompanyAccountSyncJob extends PointJob {
public class CompanyAccountSyncJob implements SimpleJob {
private PropertiesUtil propertiesUtil = PropertiesUtil.getInstance("const");
private OkHttpClient client = OkHttpUtil.getInstance();
@Override
public void executeJob(ShardingContext shardingContext, JobPointLog jobPointLog) {
public void execute(ShardingContext shardingContext) {
Request request = new Request.Builder()
.url(propertiesUtil.getPropertyAsString("COMPANY_ACCOUNT_URL"))
.get()
.addHeader("Content-Type", "application/json")
.build();
Response response = null;
String resultStr = null;
try {
response = client.newCall(request).execute();
resultStr = response.body().string();
} catch (IOException e) {
throw new RuntimeException("调用公司账户授权信息接口失败", e);
} finally {
if (response != null) {
response.close();
}
}
if (StringUtils.isNotBlank(resultStr)) {
try (Response response = client.newCall(request).execute();
SqlSession sqlSession = SessionUtil.getFactory().openSession(true)) {
DcBaseCompanyAccountMapper dcBaseCompanyAccountMapper = sqlSession.getMapper(DcBaseCompanyAccountMapper.class);
String resultStr = response.body().string();
AccountResult<CompanyAccountInfo> companyAccountInfoResult = JSONObject.parseObject(resultStr, new TypeReference<AccountResult<CompanyAccountInfo>>() {
});
if (companyAccountInfoResult.getSuccess() != null && companyAccountInfoResult.getSuccess()) {
List<CompanyAccountInfo> result = companyAccountInfoResult.getData();
if (result != null && result.size() > 0) {
DcBaseCompanyAccountMapper mapper = SessionUtil.getSession().getMapper(DcBaseCompanyAccountMapper.class);
DcBaseCompanyAccount dcBaseCompanyAccount = new DcBaseCompanyAccount();
try {
// Add by Allan at 20200718
mapper.updatesetstatusoff();
/*同步前禁用所有帐号*/
dcBaseCompanyAccountMapper.updateSetStatusOff();
for (CompanyAccountInfo companyAccountInfo : companyAccountInfoResult.getData()) {
DcBaseCompanyAccount dcBaseCompanyAccount = dcBaseCompanyAccountMapper.selectOneByExample(DcBaseCompanyAccountExample.newAndCreateCriteria()
.andAccountIdEqualTo(companyAccountInfo.getAccountId())
.andCompanyIdEqualTo(companyAccountInfo.getCompanyId())
.example());
for (CompanyAccountInfo companyAccountInfo : result) {
BeanUtils.copyProperties(dcBaseCompanyAccount, companyAccountInfo);
dcBaseCompanyAccount.setBjModified(LocalDateTime.now());
int i = mapper.updateByExampleSelective(dcBaseCompanyAccount, DcBaseCompanyAccountExample.newAndCreateCriteria()
.andCompanyIdEqualTo(companyAccountInfo.getCompanyId())
.andAccountIdEqualTo(companyAccountInfo.getAccountId()).example());
if (i == 0) {
mapper.insertSelective(dcBaseCompanyAccount);
}
}
} catch (Exception e) {
throw new RuntimeException("Mybatis操作DB失败", e);
} finally {
SessionUtil.closeSession();
}
if (dcBaseCompanyAccount == null) {
dcBaseCompanyAccount = new DcBaseCompanyAccount();
}
log.info("授权账户信息完成");
} else {
throw new RuntimeException("调用公司账户授权信息接口返回错误");
BeanUtils.copyProperties(dcBaseCompanyAccount, companyAccountInfo);
dcBaseCompanyAccount.setBjModified(LocalDateTime.now());
dcBaseCompanyAccountMapper.upsert(dcBaseCompanyAccount);
log.info("同步百伦帐号信息成功：{}", JSON.toJSONString(dcBaseCompanyAccount));
}
} else {
throw new RuntimeException("调用公司账户授权信息接口返回null");
} catch (Exception ex) {
log.error("同步百伦帐号信息成功失败:{}", ex.getMessage(), ex);
}
}
}
data-base/base-sync-company/src/test/java/CompanyTest.java
......@@ -10,18 +10,19 @@ public class CompanyTest {
@Test
public void test() {
CompanyInfoSyncJob skuSyncJob = new CompanyInfoSyncJob();
skuSyncJob.executeJob(null, new JobPointLog("",0 ,100,1,1, LocalDateTime.now().minusMonths(1),LocalDateTime.now().minusDays(29)));
skuSyncJob.executeJob(null, new JobPointLog("", 0, 100, 1, 1, LocalDateTime.now().minusMonths(1), LocalDateTime.now().minusDays(29)));
}
@Test
public void test2() {
CompanyStaffSyncJob skuSyncJob = new CompanyStaffSyncJob();
skuSyncJob.executeJob(null, new JobPointLog("",0 ,100,1,1, LocalDateTime.now().minusMonths(1),LocalDateTime.now().minusDays(29)));
skuSyncJob.executeJob(null, new JobPointLog("", 0, 100, 1, 1, LocalDateTime.now().minusMonths(1), LocalDateTime.now().minusDays(29)));
}
@Test
public void test3() {
CompanyAccountSyncJob skuSyncJob = new CompanyAccountSyncJob();
skuSyncJob.executeJob(null, new JobPointLog("",0 ,100,1,1, LocalDateTime.now().minusMonths(1),LocalDateTime.now().minusDays(29)));
skuSyncJob.execute(null);
}
}
data-base/base-sync-company/src/test/java/com/bailuntec/job/CompanyAccountSyncJobTest.java
......@@ -15,6 +15,6 @@ class CompanyAccountSyncJobTest {
@Test
void executeJob() {
CompanyAccountSyncJob companyAccountSyncJob = new CompanyAccountSyncJob();
companyAccountSyncJob.executeJob(null, null);
companyAccountSyncJob.execute(null);
}
}
data-common/pom.xml
......@@ -121,6 +121,14 @@
<artifactId>freemarker</artifactId>
</dependency>
<!--apache lang 工具包-->
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
<version>${commons-lang3.verson}</version>
</dependency>
</dependencies>
<build>
......
data-common/src/main/java/com/bailuntec/mapper/DcBaseCompanyAccountMapper.java
......@@ -122,5 +122,6 @@ public interface DcBaseCompanyAccountMapper {
*/
int upsertSelective(DcBaseCompanyAccount record);
int updatesetstatusoff();
}
\ No newline at end of file
/*同步前禁用所有帐号*/
int updateSetStatusOff();
}
data-common/src/main/java/com/bailuntec/mapper/DcBaseCostFirstMapper.java
......@@ -130,4 +130,14 @@ public interface DcBaseCostFirstMapper {
int updateHasCalculation(CostFirstOrder costFirstOrder);
}
\ No newline at end of file
void upsertBatch(@Param("lists") List<DcBaseCostFirst> dcBaseCostFirstList);
/**
* 未计算的基础头程费用
*
* @param pageStart 分页参数
* @param pageOffset 分页参数
* @return 未计算的基础头程费用
*/
List<DcBaseCostFirst> listUnCalc(@Param("pageStart") Integer pageStart, @Param("pageOffset") Integer pageOffset);
}
data-common/src/main/java/com/bailuntec/mapper/DcMidCostFirstOrderMapper.java
......@@ -123,10 +123,12 @@ public interface DcMidCostFirstOrderMapper {
*/
int upsertSelective(DcMidCostFirstOrder record);
int upsertBatch(@Param("lists") List<DcMidCostFirstOrder> list);
int countSkuWarehouse();
List<MidCostFirst> listSkuWarehouse(@Param("v1") int v1, @Param("v2") int v2);
int updateHasCalculation(MidCostFirst midCostFirst);
}
\ No newline at end of file
}
data-common/src/main/java/com/bailuntec/mapper/JobAmazonAdLogMapper.java
......@@ -2,8 +2,10 @@ package com.bailuntec.mapper;
import com.bailuntec.domain.entity.JobAmazonAdLog;
import com.bailuntec.domain.example.JobAmazonAdLogExample;
import com.dangdang.ddframe.job.api.ShardingContext;
import org.apache.ibatis.annotations.Param;
import java.time.LocalDateTime;
import java.util.List;
public interface JobAmazonAdLogMapper {
......@@ -121,4 +123,15 @@ public interface JobAmazonAdLogMapper {
* @project https://github.com/itfsw/mybatis-generator-plugin
*/
int upsertSelective(JobAmazonAdLog record);
}
\ No newline at end of file
/**
* <p>
* 获取还未下载的报告
* </p>
*
* @param shardingContext 分片参数
* @param localDateTime 获取这个日期前的报告
* @return 未下载的报告
*/
List<JobAmazonAdLog> selectUnDownload(@Param("shardingContext") ShardingContext shardingContext, @Param("localDateTime") LocalDateTime localDateTime);
}
data-common/src/main/java/com/bailuntec/utils/ListUtil.java 0 → 100644
package com.bailuntec.utils;
import org.apache.commons.collections.ListUtils;
import java.util.Collection;
import java.util.Optional;
import java.util.function.Function;
import java.util.stream.Collectors;
/**
* 集合工具类 扩展org.apache.commons.collections
*
* @author robbendev
*/
public class ListUtil extends ListUtils {
/**
* 校验集合是否为空
*
* @param coll 入参
* @return boolean
*/
public static boolean isEmpty(Collection<?> coll) {
return (coll == null || coll.isEmpty());
}
/**
* 校验集合是否不为空
*
* @param coll 入参
* @return boolean
*/
public static boolean isNotEmpty(Collection<?> coll) {
return !isEmpty(coll);
}
/**
* 根据对象属性判断是否存在
*
* @param coll 集合
* @param func 属性方法
* @param column 判断的值
* @param <T>
* @param <R>
* @return
*/
public static <T, R> boolean containsCol(Collection<T> coll, Function<T, R> func, R column) {
return coll.stream()
.map(func)
.collect(Collectors.toSet())
.contains(column);
}
public static <T, R> boolean containsObj(Collection<T> coll, Function<T, R> func, T obj) {
return coll.stream()
.map(func)
.collect(Collectors.toSet())
.contains(func.apply(obj));
}
public static <T, R> Optional<T> findOne(Collection<T> coll, Function<T, R> func, R column) {
return coll.stream()
.filter(var -> func.apply(var).equals(column))
.findFirst();
}
public static <T, R> T getOne(Collection<T> coll, Function<T, R> func, R column) {
return coll.stream()
.filter(var -> func.apply(var).equals(column))
.findFirst()
.orElse(null);
}
}
data-common/src/main/resources/logback.xml
......@@ -24,7 +24,7 @@
   
<logger name="com.bailuntec" level="DEBUG"/>
<logger name="com.bailuntec" level="INFO"/>
<logger name="org.apache.zookeeper" level="WARN"/>
<logger name="org.apache.curator" level="WARN"/>
<logger name="org.springframework" level="INFO"/>
......
data-mid/mid-duration-stock/src/main/java/com/bailuntec/job/CalculateStockRecordJob.java
......@@ -25,7 +25,7 @@ public class CalculateStockRecordJob extends PointJob {
@Override
public void executeJob(ShardingContext shardingContext, JobPointLog jobPointLog) {
jobPointLog.setPageIndex(jobPointLog.getPageIndex() != 1 ? 1 : jobPointLog.getPageIndex());
jobPointLog.setPageIndex(1);
Integer pickSize = 0;
try {
// 根据仓库、sku分组分页查询sku数量
......
data-mid/mid-duration-stock/src/test/java/com/bailuntec/job/CalculateStockRecordJobTest.java 0 → 100644
package com.bailuntec.job;
import com.bailuntec.domain.entity.JobPointLog;
import org.junit.jupiter.api.Test;
/**
* <p>
*
* </p>
*
* @author robbendev
* @since 2020/8/21 10:10 上午
*/
public class CalculateStockRecordJobTest {
@Test
public void executeJob() {
CalculateStockRecordJob job = new CalculateStockRecordJob();
JobPointLog jobPointLog = new JobPointLog();
jobPointLog.setPageIndex(1);
jobPointLog.setPageSize(100);
job.executeJob(null, jobPointLog);
}
}
data-parent/pom.xml
......@@ -54,6 +54,7 @@
<spring-cloud.version>Finchley.SR4</spring-cloud.version>
<freemarker.version>2.3.28</freemarker.version>
<curator.version>2.10.0</curator.version>
<commons-lang3.verson>3.10</commons-lang3.verson>
</properties>
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment