import java.util.regex.Matcher;
import java.util.regex.Pattern;
-
public class TaskThread implements Runnable {
- public Log log = LogFactory.getLog(TaskThread.class);
- public MessageChannel pmResultChannel;
- private MessageChannel cmResultChannel;
- private CollectMsg data;
-
- private ConfigurationInterface configurationInterface = new ConfigurationImp();
-
- private String localPath = Constant.SYS_DATA_TEMP;
- private String resultPath = Constant.SYS_DATA_RESULT;
-
- private SimpleDateFormat dateFormat = new SimpleDateFormat("yyyyMMddHHmmss");
-
- private SimpleDateFormat dateFormat2 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
-
-// private String csvpathAndFileName;
-// private String xmlPathAndFileName;
-// private int countNum = 0 ;
-
- public TaskThread(CollectMsg data) {
- this.data = data;
- }
-
- public TaskThread() {
- super();
- }
-
- @Override
- public void run() {
-
- cmResultChannel = MessageChannelFactory.getMessageChannel(Constant.COLLECT_RESULT_CHANNEL_KEY);
- pmResultChannel = MessageChannelFactory.getMessageChannel(Constant.COLLECT_RESULT_PM_CHANNEL_KEY);
- try {
- collectMsgHandle(data);
- } catch (Exception e) {
- log.error("", e);
- }
- }
-
- private void collectMsgHandle(CollectMsg collectMsg) {
- String emsName = collectMsg.getEmsName();
- String type = collectMsg.getType();
- CollectVo collectVo = configurationInterface.getCollectVoByEmsNameAndType(emsName, type);
-
- //ftp download
- List<String> downloadfiles = this.ftpDownload(collectVo);
- //paser ftp update message send
- for (String fileName : downloadfiles) {
- this.parseFtpAndSendMessage(fileName, collectVo);
- }
- }
-
- public void parseFtpAndSendMessage(String fileName, CollectVo collectVo) {
- //
- List<File> filelist = decompressed(fileName);
-
- for (File tempfile : filelist) {
-
- String unfileName = tempfile.getName();
-
- Pattern pa = Pattern.compile(".*-(.*)-\\w{2}-");
- Matcher ma = pa.matcher(unfileName);
- if (!ma.find())
- continue;
- String nename = ma.group(1);
- boolean parseResult = false;
- if (Constant.COLLECT_TYPE_CM.equalsIgnoreCase(collectVo.getType())) {
- parseResult = processCMXml(tempfile, nename, "CM");
- } else {
- if (unfileName.indexOf(".csv") > 0) {
- parseResult = processPMCsv(tempfile);
- } else {
- parseResult = processPMXml(tempfile);
- }
- }
-
- if (parseResult) {
- log.info("parser " + tempfile + " sucess");
- tempfile.delete();
- } else {
- log.info("parser " + tempfile + " fail");
- }
-
- }
- }
-
- public boolean processPMXml(File file) {
-
- FileInputStream fis = null;
- InputStreamReader isr = null;
- XMLStreamReader reader = null;
- try {
- fis = new FileInputStream(file);
- isr = new InputStreamReader(fis, Constant.ENCODING_UTF8);
-
- XMLInputFactory fac = XMLInputFactory.newInstance();
- reader = fac.createXMLStreamReader(isr);
-
- boolean fileHeaderStart = false;
- boolean measurementStart = false;
- boolean pmNameFlag = false;
- boolean pmDataFlag = false;
- boolean objectFlag = true;
-
- int index = -1;
- int nameIndex = -1;
- String currentMea = null;
- String subName = null;
- String localName = null;
- String endLocalName = null;
- String objectType = null;
-
-
- LinkedHashMap<String, String> commonNameAndValue = new LinkedHashMap<String, String>();
- LinkedHashMap<String, String> pmDatas = null;
- LinkedHashMap<Integer, String> pmNames = null;
-
-
- int event = -1;
- while (reader.hasNext()) {
- try {
- event = reader.next();
-
- switch (event) {
- case XMLStreamConstants.START_ELEMENT:
- localName = reader.getLocalName();
- if ("FileHeader".equalsIgnoreCase(localName)) {
- fileHeaderStart = true;
- }
- if (fileHeaderStart) {
- if (!"FileHeader".equalsIgnoreCase(localName)) {
- commonNameAndValue.put(localName, reader.getElementText().trim());
- }
-
- }
- if ("Measurements".equalsIgnoreCase(localName)) {
- // a new Measurement starts
- measurementStart = true;
- }
- if (measurementStart) {
- // measurement handler
- if ("ObjectType".equalsIgnoreCase(localName)) {
- objectType = reader.getElementText().trim();
- commonNameAndValue.put("ObjectType", objectType);
- }
- if ("PmName".equalsIgnoreCase(localName)) {
- pmNameFlag = true;
- pmNames = new LinkedHashMap<Integer, String>();
-
- }
- if (pmNameFlag) {
- // pmname handler, add columnNames
- if ("N".equalsIgnoreCase(localName)) {
- nameIndex = Integer.parseInt(getXMLAttribute(reader, "i"));
- String text = reader.getElementText().trim();
- pmNames.put(nameIndex, text);
- }
- }
- if ("PmData".equalsIgnoreCase(localName)) {
- pmDataFlag = true;
- pmDatas = new LinkedHashMap<String, String>();
- }
-
- if (pmDataFlag) {
- // pmdata handler
- if ("Object".equalsIgnoreCase(localName)) {
- objectFlag = true;
- int n = reader.getAttributeCount();
- for (int i = 0; i < n; i++) {
- String name = reader.getAttributeLocalName(i);
- commonNameAndValue.put(name, reader.getAttributeValue(i));
- }
- }
- if (objectFlag) {
-
- // add columnValues
- if ("V".equalsIgnoreCase(localName)) {
- String indexStr = getXMLAttribute(reader, "i");
- if (indexStr == null) {
- log.error("ERROR: illegal value index");
- continue;
- }
- index = Integer.parseInt(indexStr);
- String name = pmNames.get(index);
- if (name == null) {
- log.error("illegal data: valueIndex=" + index);
- continue;
- }
-
- String value = reader.getElementText().trim();
- pmDatas.put(name, value);
- }
- if ("CV".equalsIgnoreCase(localName)) {
-
- String indexStr = getXMLAttribute(reader, "i");
- if (indexStr == null) {
- log.error("ERROR: illegal value index");
- continue;
- }
- index = Integer.parseInt(indexStr);
-
- currentMea = pmNames.get(index);
- if (currentMea == null) {
- log.error("illegal data: valueIndex=" + index);
- continue;
- }
- }
-
- if ("SN".equalsIgnoreCase(localName)) {
- subName = reader.getElementText().trim();
-
- }
- if ("SV".equalsIgnoreCase(localName)) {
- String subValue = reader.getElementText().trim();
-// pmDatas.put(currentMea+subName, subValue);
- pmDatas.put(subName, subValue);
- }
- }
- }
-
- }
-
- break;
- case XMLStreamConstants.CHARACTERS:
- // ...
- break;
- case XMLStreamConstants.END_ELEMENT:
- // ...
- endLocalName = reader.getLocalName();
- if ("Object".equalsIgnoreCase(endLocalName)) {
- objectFlag = false;
- pmDatas.putAll(commonNameAndValue);
- try {
- pmResultChannel.put(pmDatas);
-
- } catch (InterruptedException e) {
- log.error("collectResultChannel.put(resultMap) error ", e);
- }
-// System.out.println(pmDatas);
-// pmDatas.clear();
- }
- if (endLocalName.equalsIgnoreCase("PmData")) {
- pmDataFlag = false;
- }
-
- if (endLocalName.equalsIgnoreCase("PmName")) {
- pmNameFlag = false;
- }
- if (endLocalName.equalsIgnoreCase("Measurements")) {
- // a measurement over
- measurementStart = false;
- }
-
- if ("FileHeader".equalsIgnoreCase(endLocalName)) {
- fileHeaderStart = false;
- }
- break;
- }
- } catch (Exception e) {
- log.error("", e);
- event = reader.next();
- }
- }
-
- } catch (Exception e) {
- log.error("processPMXml is Exception ", e);
- return false;
- } finally {
- try {
- if (reader != null) reader.close();
- if (isr != null) isr.close();
- if (fis != null) fis.close();
- } catch (Exception e) {
- e.printStackTrace();
- }
- }
- return true;
- }
-
- private String getXMLAttribute(XMLStreamReader reader, String obj) {
- String res = null;
- if (obj == null || reader == null) {
- return res;
- }
- int n = reader.getAttributeCount();
- for (int i = 0; i < n; i++) {
- String name = reader.getAttributeLocalName(i);
- if (obj.equalsIgnoreCase(name)) {
- res = reader.getAttributeValue(i);
- }
- }
- return res;
- }
-
- public boolean processPMCsv(File tempfile) {
-
- FileInputStream brs = null;
- InputStreamReader isr = null;
- BufferedReader br = null;
-
- List<String> columnNames = new ArrayList<String>();
- List<String> commonValues = new ArrayList<String>();
- try {
-
- brs = new FileInputStream(tempfile);
- isr = new InputStreamReader(brs, Constant.ENCODING_UTF8);
- br = new BufferedReader(isr);
- //common field
- String commonField = br.readLine();
- String[] fields = commonField.split("\\|", -1);
- for (String com : fields) {
- String[] comNameAndValue = com.split("=", 2);
- columnNames.add(comNameAndValue[0].trim());
- commonValues.add(comNameAndValue[1]);
- }
- //column names
- String columnName = br.readLine();
- String[] names = columnName.split("\\|", -1);
- for (String name : names) {
- columnNames.add(name);
- }
-
- String valueLine = "";
- List<String> valuelist = new ArrayList<String>();
-
- while ((valueLine = br.readLine()) != null) {
- if (valueLine.trim().equals("")) {
- continue;
- }
-// countNum ++;
- String[] values = valueLine.split("\\|", -1);
-
- valuelist.addAll(commonValues);
- for (String value : values) {
- valuelist.add(value);
- }
-// this.appendLine(valuelist, bos);
- //resultMap
- HashMap<String, String> resultMap = this.resultMap(columnNames, valuelist);
- try {
- pmResultChannel.put(resultMap);
- } catch (InterruptedException e) {
- log.error("collectResultChannel.put(resultMap) error ", e);
- }
- valuelist.clear();
- }
- } catch (IOException e) {
- log.error("processPMCsv is fail ", e);
- return false;
- } finally {
- try {
- if (br != null)
- br.close();
- if (isr != null)
- isr.close();
- if (brs != null)
- brs.close();
-
- } catch (Exception e) {
- log.error(e);
- }
- }
- return true;
-
- }
-
- private HashMap<String, String> resultMap(List<String> columnNames, List<String> valuelist) {
-
- HashMap<String, String> resultMap = new HashMap<String, String>();
- if (columnNames.size() == valuelist.size()) {
- for (int i = 0; i < columnNames.size(); i++) {
- resultMap.put(columnNames.get(i), valuelist.get(i));
- }
- }
-
- return resultMap;
-
- }
-
- private boolean processCMXml(File tempfile, String nename, String type) {
-
- String csvpath = localPath + nename + "/" + type + "/";
- File csvpathfile = new File(csvpath);
- if (!csvpathfile.exists()) {
- csvpathfile.mkdirs();
- }
- String csvFileName = nename + dateFormat.format(new Date()) + System.nanoTime();
- String csvpathAndFileName = csvpath + csvFileName + ".csv";
- BufferedOutputStream bos = null;
- FileOutputStream fos = null;
- try {
- fos = new FileOutputStream(csvpathAndFileName, false);
- bos = new BufferedOutputStream(fos, 10240);
- } catch (FileNotFoundException e1) {
- log.error("FileNotFoundException " + StringUtil.getStackTrace(e1));
- }
-
- boolean FieldNameFlag = false;
- boolean FieldValueFlag = false;
- //line num
- int countNum = 0;
- String xmlPathAndFileName = null;
- String localName = null;
- String endLocalName = null;
- String rmUID = null;
- int index = -1;
- ArrayList<String> names = new ArrayList<String>();// colname
- LinkedHashMap<String, String> nameAndValue = new LinkedHashMap<String, String>();
-
-
- FileInputStream fis = null;
- InputStreamReader isr = null;
- XMLStreamReader reader = null;
- try {
- fis = new FileInputStream(tempfile);
- isr = new InputStreamReader(fis, Constant.ENCODING_UTF8);
- XMLInputFactory fac = XMLInputFactory.newInstance();
- reader = fac.createXMLStreamReader(isr);
- int event = -1;
- boolean setcolum = true;
- while (reader.hasNext()) {
- try {
- event = reader.next();
- switch (event) {
- case XMLStreamConstants.START_ELEMENT:
- localName = reader.getLocalName();
- if ("FieldName".equalsIgnoreCase(localName)) {
- FieldNameFlag = true;
- }
- if (FieldNameFlag) {
- if ("N".equalsIgnoreCase(localName)) {
- String colName = reader.getElementText().trim();
- names.add(colName);
- }
- }
- if ("FieldValue".equalsIgnoreCase(localName)) {
- FieldValueFlag = true;
-
- }
- if (FieldValueFlag) {
- if (setcolum) {
- xmlPathAndFileName = this.setColumnNames(nename, names, type);
- setcolum = false;
- }
-
- if ("Object".equalsIgnoreCase(localName)) {
- int ac = reader.getAttributeCount();
- for (int i = 0; i < ac; i++) {
- if ("rmUID".equalsIgnoreCase(reader.getAttributeLocalName(i))) {
- rmUID = reader.getAttributeValue(i).trim();
- }
- }
- nameAndValue.put("rmUID", rmUID);
- }
- if ("V".equalsIgnoreCase(localName)) {
- index = Integer.parseInt(reader
- .getAttributeValue(0)) - 1;
- String currentName = names.get(index);
- String v = reader.getElementText().trim();
- nameAndValue.put(currentName, v);
- }
- }
- break;
- case XMLStreamConstants.CHARACTERS:
- break;
- case XMLStreamConstants.END_ELEMENT:
- endLocalName = reader.getLocalName();
-
- if ("FieldName".equalsIgnoreCase(endLocalName)) {
- FieldNameFlag = false;
- }
- if ("FieldValue".equalsIgnoreCase(endLocalName)) {
- FieldValueFlag = false;
- }
- if ("Object".equalsIgnoreCase(endLocalName)) {
- countNum++;
- this.appendLine(nameAndValue, bos);
- nameAndValue.clear();
- }
- break;
- }
- } catch (Exception e) {
- log.error("" + StringUtil.getStackTrace(e));
- event = reader.next();
- }
- }
-
-
- if (bos != null) {
- bos.close();
- bos = null;
- }
- if (fos != null) {
- fos.close();
- fos = null;
- }
-
- String[] fileKeys = this.createZipFile(csvpathAndFileName, xmlPathAndFileName, nename);
- //ftp store
- Properties ftpPro = configurationInterface.getProperties();
- String ip = ftpPro.getProperty("ftp_ip");
- String port = ftpPro.getProperty("ftp_port");
- String ftp_user = ftpPro.getProperty("ftp_user");
- String ftp_password = ftpPro.getProperty("ftp_password");
-
- String ftp_passive = ftpPro.getProperty("ftp_passive");
- String ftp_type = ftpPro.getProperty("ftp_type");
- String remoteFile = ftpPro.getProperty("ftp_remote_path");
- this.ftpStore(fileKeys, ip, port, ftp_user, ftp_password, ftp_passive, ftp_type, remoteFile);
- //create Message
- String message = this.createMessage(fileKeys[1], ftp_user, ftp_password, ip, port, countNum, nename);
-
- //set message
- this.setMessage(message);
- } catch (Exception e) {
- log.error("" + StringUtil.getStackTrace(e));
- return false;
- } finally {
- try {
- if (reader != null) {
- reader.close();
- }
- if (isr != null) {
- isr.close();
- }
- if (fis != null) {
- fis.close();
- }
- if (bos != null) {
- bos.close();
- }
-
- if (fos != null) {
- fos.close();
- }
- } catch (Exception e) {
- log.error(e);
- }
- }
- return true;
- }
-
- private void setMessage(String message) {
-
- try {
- cmResultChannel.put(message);
- } catch (Exception e) {
- log.error("collectResultChannel.put(message) is error " + StringUtil.getStackTrace(e));
- }
- }
-
- public String createMessage(String zipName, String user, String pwd, String ip, String port, int countNum, String nename) {
-
- StringBuffer strBuffer = new StringBuffer();
- strBuffer
- .append("<?xml version=\"1.0\" encoding=\"UTF-8\"?>"
- + "<FILE_DATA_READY_UL xmlns:xsi=\" http://www.w3.org/2001/XMLSchema-instance\">"
- + "<Header SessionID=\"");
- strBuffer.append("");
- strBuffer.append("\" LicenceID=\"");
- strBuffer.append("");
- strBuffer.append("\" SystemID=\"");
- strBuffer.append("");
- strBuffer.append("\" Time=\"");
- strBuffer.append(dateFormat2.format(new Date()));
- strBuffer.append("\" PolicyID=\"");
- strBuffer.append("");
- strBuffer.append("\"/><Body>");
- strBuffer.append("<DataCatalog>");
- strBuffer.append("");
- strBuffer.append("</DataCatalog><GroupID>");
- strBuffer.append(nename);
- strBuffer.append("</GroupID><DataSourceName>");
- strBuffer.append("");
- strBuffer.append("</DataSourceName><InstanceID>");
- strBuffer.append("");
- strBuffer.append("</InstanceID><FileFormat>");
- strBuffer.append("csv");
- strBuffer.append("</FileFormat><CharSet>");
- strBuffer.append("gbk");
- strBuffer.append("</CharSet><FieldSeparator>");
- strBuffer.append("|");
- strBuffer.append("</FieldSeparator><IsCompressed>");
- strBuffer.append("true");
- strBuffer.append("</IsCompressed><StartTime>");
- strBuffer.append(dateFormat2.format(new Date()));
- strBuffer.append("</StartTime><EndTime>");
- strBuffer.append("");
- strBuffer.append("</EndTime><FileList>");
- strBuffer.append(zipName);
- strBuffer.append("</FileList><ConnectionString>");
- strBuffer.append("ftp://" + user + ":" + pwd + "@" + ip + ":" + port);
- strBuffer.append("</ConnectionString>");
- strBuffer.append("<DataCount>");
- strBuffer.append(countNum);
- strBuffer.append("</DataCount>");
-
- strBuffer.append("<FileSize>").append("").append("</FileSize>");
- strBuffer.append("<DataGranularity>").append("").append("</DataGranularity>");
-
-
- strBuffer.append("</Body></FILE_DATA_READY_UL>");
- return strBuffer.toString();
-
- }
-
- private void ftpStore(String[] fileKeys, String ip, String port, String ftp_user, String ftp_password,
- String ftp_passive, String ftp_type, String remoteFile) {
- String zipFilePath = fileKeys[0];
-
-
- FTPInterface ftpClient;
- ftpClient = new FTPSrv();
- //login
- try {
- ftpClient.login(ip, Integer.parseInt(port), ftp_user, ftp_password, "GBK", Boolean.parseBoolean(ftp_passive), 5 * 60 * 1000);
- } catch (Exception e) {
- log.error("login fail,ip=[" + ip + "] port=[" + port + "] user=[" + ftp_user + "]pwd=[" + ftp_password + "]" + StringUtil.getStackTrace(e));
- return;
- }
-// ftpClient.store(zipFilePath, remoteFile);
- log.debug("store [" + zipFilePath + "]to[" + remoteFile + "]");
-
- FileUtils.deleteQuietly(new File(zipFilePath));
-
-
- }
-
- private String[] createZipFile(String csvpathAndFileName, String xmlPathAndFileName, String nename) {
-
- String zipPath = resultPath + nename + dateFormat.format(new Date()) + "_" + System.nanoTime();
-
- File destDir = new File(zipPath);
- destDir.mkdirs();
-
- try {
- FileUtils.copyFileToDirectory(new File(csvpathAndFileName), destDir);
- FileUtils.copyFileToDirectory(new File(xmlPathAndFileName), destDir);
- } catch (IOException e) {
-
- }
-
- String destFilePath = zipPath + ".zip";
- try {
- Zip zip = new Zip(destDir.getAbsolutePath(), destFilePath);
- zip.setCompressLevel(9);
- zip.compress();
-
- FileUtils.deleteDirectory(destDir);
- } catch (IOException e) {
- log.error("zip.compress() is fail " + StringUtil.getStackTrace(e));
- }
- return new String[]{destFilePath, zipPath + ".zip"};
- }
-
-
- private String setColumnNames(String nename, List<String> names, String type) {
- //write xml
- String xmlpath = localPath + nename + "/" + type + "/";
- File xmlpathfile = new File(xmlpath);
- if (!xmlpathfile.exists()) {
- xmlpathfile.mkdirs();
- }
- String xmlFileName = nename + dateFormat.format(new Date()) + System.nanoTime();
- String fieldLine = "";
- for (int i = 0; i < names.size(); i++) {
- String field = "\t<Field>\r\n" + "\t\t<FieldNo>" + i
- + "</FieldNo>\r\n" + "\t\t<FieldName>"
- + names.get(i) + "</FieldName>\r\n"
- + "\t\t<FieldType>" + names.get(i)
- + "</FieldType>\r\n"
- + "\t\t<FieldNameOther>" + names.get(i)
- + "</FieldNameOther>\r\n" +
- "\t</Field>\r\n";
- fieldLine = fieldLine + field;
- }
-
- String str = "<?xml version=\"1.0\" encoding=\"gbk\"?>\r\n"
- + "<xml>\r\n" + "<FILE_STRUCTURE>\r\n" + fieldLine
- + "</FILE_STRUCTURE>\r\n" + "</xml>\r\n";
- String xmlPathAndFileName = xmlpath + xmlFileName + ".xml";
- try {
- this.writeDetail(xmlPathAndFileName, str);
- } catch (Exception e) {
- log.error("writeDetail is fail ,xmlFileName=" + xmlFileName + StringUtil.getStackTrace(e));
- }
-
- return xmlPathAndFileName;
- }
-
- private void writeDetail(String detailFileName, String str) throws Exception {
- OutputStreamWriter writer = null;
- OutputStream readOut = null;
- try {
- readOut = new FileOutputStream(new File(detailFileName), false);
- writer = new OutputStreamWriter(readOut);
- writer.write(str);
- writer.flush();
- } finally {
-
- if (null != writer) {
- writer.close();
- }
- if (readOut != null) {
- readOut.close();
- }
-
- }
-
- }
-
-
- private void appendLine(LinkedHashMap<String, String> nameAndValue, BufferedOutputStream bos) {
- StringBuilder lineDatas = new StringBuilder();
-
- for (String key : nameAndValue.keySet()) {
- lineDatas.append(nameAndValue.get(key)).append("|");
- }
- try {
- bos.write(lineDatas.toString().getBytes());
- bos.write("\n".getBytes());
- } catch (IOException e) {
- log.error("appendLine error " + StringUtil.getStackTrace(e));
- }
- }
-
-// private void appendLine(List<String> values,BufferedOutputStream bos) {
-// StringBuilder lineDatas = new StringBuilder();
-//
-// for (String value : values) {
-// lineDatas.append(value).append("|");
-// }
-// try {
-// bos.write(lineDatas.toString().getBytes());
-// bos.write("\n".getBytes());
-// } catch (IOException e) {
-// log.error("appendLine error "+StringUtil.getStackTrace(e));
-// }
-// }
-
- public List<File> decompressed(String fileName) {
- List<File> filelist = new ArrayList<File>();
-
- if (fileName.indexOf(".gz") > 1) {
- try {
- File decompressFile = deGz(fileName);
- filelist.add(decompressFile);
- new File(fileName).delete();
- } catch (IOException e) {
- log.error("decompressed is fail " + StringUtil.getStackTrace(e));
- }
- } else if (fileName.indexOf(".zip") > 1) {
- try {
- File[] files = deZip(new File(fileName));
- new File(fileName).delete();
- for (File temp : files) {
- filelist.add(temp);
- }
- } catch (Exception e) {
- log.error("decompressed is fail " + StringUtil.getStackTrace(e));
- }
- } else {
- filelist.add(new File(fileName));
- }
-
- return filelist;
- }
-
- private File deGz(String gzFileName) throws IOException {
- Gunzip gunzip = new Gunzip();
- String orgFile = gzFileName.replace(".gz", "");
- gunzip.unCompress(gzFileName, orgFile);
- return new File(orgFile);
- }
-
- public File[] deZip(File file) throws Exception {
-
- String regx = "(.*).zip";
- Pattern p = Pattern.compile(regx);
- Matcher m = p.matcher(file.getName());
- if (m.find()) {
- String orgFile = localPath + m.group(1) + "/";
- UnZip unzip = new UnZip(file.getAbsolutePath(), orgFile);
- unzip.deCompress();
- file = new File(orgFile);
- }
- File[] files = file.listFiles();
-
- return files;
-
- }
-
- private List<String> ftpDownload(CollectVo collectVo) {
-
- List<String> fileList = new ArrayList<String>();
- //IP
- String ip = collectVo.getIP();
- //port
- String port = collectVo.getPort();
- //user
- String user = collectVo.getUser();
- //password
- String password = collectVo.getPassword();
- //isPassiveMode
- String passivemode = collectVo.getPassive();
-
- FTPInterface ftpClient = new FTPSrv();
-
- //login
- try {
- log.info("ftp login ,ip=[" + ip + "] port=[" + port + "] user=[" + user + "]password=[" + password + "]");
- ftpClient.login(ip, Integer.parseInt(port), user, password, "GBK", Boolean.parseBoolean(passivemode), 5 * 60 * 1000);
- } catch (Exception e) {
- log.error("login fail,ip=[" + ip + "] port=[" + port + "] user=[" + user + "]password=[" + password + "]" + StringUtil.getStackTrace(e));
- return fileList;
- }
-
- //download
- String dir = collectVo.getRemotepath();
- List<String> searchExprList = new ArrayList<String>();
- String[] FPath = dir.split(";");
- for (int i = 0; i < FPath.length; i++) {
- int oldSize = searchExprList.size();
- String conpath = FPath[i] + collectVo.getMatch();
- Hashtable<String, String> varMap = new Hashtable<String, String>();
- long collectPeriod = 900;
- try {
- collectPeriod = Long.parseLong(collectVo.getGranularity()) * 60;
- log.info("collectPeriod =[" + collectPeriod + "]");
- } catch (NumberFormatException e) {
- e.printStackTrace();
- }
- long[] d = DateUtil.getScanScope(new Date(), collectPeriod);
- searchExprList.add(VarExprParser.replaceVar(conpath, d[0], d[1]));
-
- varMap.clear();
- varMap = null;
- log.info("[" + conpath + "] result[" + (searchExprList.size() - oldSize) + "] path");
- conpath = null;
- }
- String nowdir = null;
- try {
- nowdir = ftpClient.pwd();
- searchExprList = getPathNoRegular(searchExprList, ftpClient);
- } catch (Exception e1) {
- log.error(" collect fail ", e1);
- return fileList;
- }
- List<AFtpRemoteFile> remoteFiles = new ArrayList<AFtpRemoteFile>();
- for (String expr : searchExprList) {
- ftpClient.chdir(nowdir);
- String keys[] = parseExprKeys(expr);
- String ftpRegular = keys[1];
- String ftpDir = keys[0];
-
- boolean cdsucess = ftpClient.chdir(ftpDir);
- if (cdsucess) {
- AFtpRemoteFile[] arf = (AFtpRemoteFile[]) ftpClient.list();
- log.info(" list [" + ftpDir + "] result[" + (arf == null ? "null" : arf.length) + "] files");
- //filter
-
- rfileFilter(remoteFiles, arf, ftpRegular);
-
- keys = null;
- ftpRegular = ftpDir = null;
-
- for (AFtpRemoteFile ftpRemoteFile : remoteFiles) {
- if (!new File(localPath).exists()) {
- try {
- new File(localPath).mkdir();
- } catch (Exception e) {
- log.error("create localPath is fail localPath=" + localPath + " " + StringUtil.getStackTrace(e));
- }
- }
-
- if (!new File(localPath).exists()) {
- new File(localPath).mkdirs();
- }
-
- String localFileName = localPath + ftpRemoteFile.getFileName();
- File loaclFile = new File(localFileName);
- if (loaclFile.exists()) {
- loaclFile.delete();
- }
-
- boolean flag = ftpClient.downloadFile(ftpRemoteFile.getAbsFileName(), localFileName);
-
- if (flag) {
- fileList.add(localFileName);
- } else {
- log.error("download file fail fileName=" + ftpRemoteFile.getAbsFileName());
- }
- }
-
- } else {
- log.error("cd dir is faill dir =[" + ftpDir + "]");
- }
- }
-
-
- return fileList;
- }
-
- private void rfileFilter(List<AFtpRemoteFile> fileContainer, AFtpRemoteFile[] arfs, String ftpRegular) {
- if (ftpRegular != null && ftpRegular.length() > 0) {
- Pattern pattern = null;
- try {
- pattern = Pattern.compile(ftpRegular, Pattern.CASE_INSENSITIVE);
- } catch (Exception e) {
- log.info("[" + ftpRegular + "]Pattern.compile exception:" + e.getMessage());
- //should rethrow exception or return from here
- }
- int hisSize = fileContainer.size();
- for (int j = 0; arfs != null && j < arfs.length; j++) {
- String fileName = parseFileName(arfs[j].getFileName());
- Matcher matcher = null;
- if (pattern != null)
- matcher = pattern.matcher(fileName);
- else {
- //define the flow when pattern is null
- }
-
- if (matcher.find())
- fileContainer.add(arfs[j]);
- }
- log.info("[" + ftpRegular + "]filter[" + (fileContainer.size() - hisSize) + "]filse");
- pattern = null;
- } else {
- for (int j = 0; arfs != null && j < arfs.length; j++)
- fileContainer.add(arfs[j]);
- }
-
- }
-
- private String parseFileName(String fileName) {
- int idx = fileName.lastIndexOf("/");
- if (idx == -1)
- return fileName;
- return fileName.substring(idx + 1, fileName.length());
- }
-
- private String[] parseExprKeys(String source) {
-
- if (source.indexOf(";") > -1) {
- source = source.substring(0, source.indexOf(";"));
- }
- if (source.endsWith("/"))
- return new String[]{source, ""};
-
- int idx = source.lastIndexOf("/");
- String[] dirkeys = new String[2];
- dirkeys[0] = source.substring(0, idx + 1);
- dirkeys[1] = source.substring(idx + 1, source.length());
- return dirkeys;
- }
-
-
- public List<String> getPathNoRegular(List<String> searchExprList, FTPInterface ftpCache) throws Exception {
- boolean isregular = false;
- List<String> regularList = new ArrayList<String>();
- for (String regular : searchExprList) {
- Pattern lpattern = null;
- try {
- lpattern = Pattern.compile("(.*/)<([^/]+)>(/.*)");
- } catch (Exception e) {
- log.error("[" + regular + "]compile fails:" + e.getMessage());
- e.printStackTrace();
- }
- Matcher matcher=null;
- if(lpattern!=null)
- matcher = lpattern.matcher(regular);
- else{
- //define flow in case lpattern is null
- }
-
- if (matcher.find()) {
- isregular = true;
- String parpath = matcher.group(1);
- try {
- boolean isin = ftpCache.chdir(parpath);
- if (isin) {
- log.info("cd dir [" + parpath + "] sucess");
- } else {
- log.error("cd dir [" + parpath + "] fail");
- }
- } catch (Exception e) {
- log.error(" cd dir [" + parpath + "]fail", e);
- throw e;
- }
- RemoteFile[] remotef = ftpCache.list();
- for (RemoteFile aremote : remotef) {
- if (aremote.isDirectory() && aremote.getFileName().matches(matcher.group(2))) {
- regularList.add(matcher.group(1) + aremote.getFileName() + matcher.group(3));
- }
- }
- } else {
- regularList.add(regular);
- }
- }
- if (isregular == true) {
- getPathNoRegular(regularList, ftpCache);
- }
- return regularList;
- }
+ public Log log = LogFactory.getLog(TaskThread.class);
+ public MessageChannel pmResultChannel;
+ private MessageChannel cmResultChannel;
+ private CollectMsg data;
+
+ private ConfigurationInterface configurationInterface = new ConfigurationImp();
+
+ private String localPath = Constant.SYS_DATA_TEMP;
+ private String resultPath = Constant.SYS_DATA_RESULT;
+
+ private SimpleDateFormat dateFormat = new SimpleDateFormat("yyyyMMddHHmmss");
+
+ private SimpleDateFormat dateFormat2 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+
+ // private String csvpathAndFileName;
+ // private String xmlPathAndFileName;
+ // private int countNum = 0 ;
+
+ public TaskThread(CollectMsg data) {
+ this.data = data;
+ }
+
+ public TaskThread() {
+ super();
+ }
+
+ @Override
+ public void run() {
+
+ cmResultChannel = MessageChannelFactory.getMessageChannel(Constant.COLLECT_RESULT_CHANNEL_KEY);
+ pmResultChannel = MessageChannelFactory.getMessageChannel(Constant.COLLECT_RESULT_PM_CHANNEL_KEY);
+ try {
+ collectMsgHandle(data);
+ } catch (Exception e) {
+ log.error("", e);
+ }
+ }
+
+ private void collectMsgHandle(CollectMsg collectMsg) {
+ String emsName = collectMsg.getEmsName();
+ String type = collectMsg.getType();
+ CollectVo collectVo = configurationInterface.getCollectVoByEmsNameAndType(emsName, type);
+
+ // ftp download
+ List<String> downloadfiles = this.ftpDownload(collectVo);
+ // paser ftp update message send
+ for (String fileName : downloadfiles) {
+ this.parseFtpAndSendMessage(fileName, collectVo);
+ }
+ }
+
+ public void parseFtpAndSendMessage(String fileName, CollectVo collectVo) {
+ //
+ List<File> filelist = decompressed(fileName);
+
+ for (File tempfile : filelist) {
+
+ String unfileName = tempfile.getName();
+
+ Pattern pa = Pattern.compile(".*-(.*)-\\w{2}-");
+ Matcher ma = pa.matcher(unfileName);
+ if (!ma.find())
+ continue;
+ String nename = ma.group(1);
+ boolean parseResult = false;
+ if (Constant.COLLECT_TYPE_CM.equalsIgnoreCase(collectVo.getType())) {
+ parseResult = processCMXml(tempfile, nename, "CM");
+ } else {
+ if (unfileName.indexOf(".csv") > 0) {
+ parseResult = processPMCsv(tempfile);
+ } else {
+ parseResult = processPMXml(tempfile);
+ }
+ }
+
+ if (parseResult) {
+ log.info("parser " + tempfile + " sucess");
+ tempfile.delete();
+ } else {
+ log.info("parser " + tempfile + " fail");
+ }
+
+ }
+ }
+
+ public boolean processPMXml(File file) {
+
+ FileInputStream fis = null;
+ InputStreamReader isr = null;
+ XMLStreamReader reader = null;
+ try {
+ fis = new FileInputStream(file);
+ isr = new InputStreamReader(fis, Constant.ENCODING_UTF8);
+
+ XMLInputFactory fac = XMLInputFactory.newInstance();
+ reader = fac.createXMLStreamReader(isr);
+
+ boolean fileHeaderStart = false;
+ boolean measurementStart = false;
+ boolean pmNameFlag = false;
+ boolean pmDataFlag = false;
+ boolean objectFlag = true;
+
+ int index = -1;
+ int nameIndex = -1;
+ String currentMea = null;
+ String subName = null;
+ String localName = null;
+ String endLocalName = null;
+ String objectType = null;
+
+ LinkedHashMap<String, String> commonNameAndValue = new LinkedHashMap<String, String>();
+ LinkedHashMap<String, String> pmDatas = null;
+ LinkedHashMap<Integer, String> pmNames = null;
+
+ int event = -1;
+ while (reader.hasNext()) {
+ try {
+ event = reader.next();
+
+ switch (event) {
+ case XMLStreamConstants.START_ELEMENT:
+ localName = reader.getLocalName();
+ if ("FileHeader".equalsIgnoreCase(localName)) {
+ fileHeaderStart = true;
+ }
+ if (fileHeaderStart) {
+ if (!"FileHeader".equalsIgnoreCase(localName)) {
+ commonNameAndValue.put(localName, reader.getElementText().trim());
+ }
+
+ }
+ if ("Measurements".equalsIgnoreCase(localName)) {
+ // a new Measurement starts
+ measurementStart = true;
+ }
+ if (measurementStart) {
+ // measurement handler
+ if ("ObjectType".equalsIgnoreCase(localName)) {
+ objectType = reader.getElementText().trim();
+ commonNameAndValue.put("ObjectType", objectType);
+ }
+ if ("PmName".equalsIgnoreCase(localName)) {
+ pmNameFlag = true;
+ pmNames = new LinkedHashMap<Integer, String>();
+
+ }
+ if (pmNameFlag) {
+ // pmname handler, add columnNames
+ if ("N".equalsIgnoreCase(localName)) {
+ nameIndex = Integer.parseInt(getXMLAttribute(reader, "i"));
+ String text = reader.getElementText().trim();
+ pmNames.put(nameIndex, text);
+ }
+ }
+ if ("PmData".equalsIgnoreCase(localName)) {
+ pmDataFlag = true;
+ pmDatas = new LinkedHashMap<String, String>();
+ }
+
+ if (pmDataFlag) {
+ // pmdata handler
+ if ("Object".equalsIgnoreCase(localName)) {
+ objectFlag = true;
+ int n = reader.getAttributeCount();
+ for (int i = 0; i < n; i++) {
+ String name = reader.getAttributeLocalName(i);
+ commonNameAndValue.put(name, reader.getAttributeValue(i));
+ }
+ }
+ if (objectFlag) {
+
+ // add columnValues
+ if ("V".equalsIgnoreCase(localName)) {
+ String indexStr = getXMLAttribute(reader, "i");
+ if (indexStr == null) {
+ log.error("ERROR: illegal value index");
+ continue;
+ }
+ index = Integer.parseInt(indexStr);
+ String name = pmNames.get(index);
+ if (name == null) {
+ log.error("illegal data: valueIndex=" + index);
+ continue;
+ }
+
+ String value = reader.getElementText().trim();
+ pmDatas.put(name, value);
+ }
+ if ("CV".equalsIgnoreCase(localName)) {
+
+ String indexStr = getXMLAttribute(reader, "i");
+ if (indexStr == null) {
+ log.error("ERROR: illegal value index");
+ continue;
+ }
+ index = Integer.parseInt(indexStr);
+
+ currentMea = pmNames.get(index);
+ if (currentMea == null) {
+ log.error("illegal data: valueIndex=" + index);
+ continue;
+ }
+ }
+
+ if ("SN".equalsIgnoreCase(localName)) {
+ subName = reader.getElementText().trim();
+
+ }
+ if ("SV".equalsIgnoreCase(localName)) {
+ String subValue = reader.getElementText().trim();
+ // pmDatas.put(currentMea+subName,
+ // subValue);
+ pmDatas.put(subName, subValue);
+ }
+ }
+ }
+
+ }
+
+ break;
+ case XMLStreamConstants.CHARACTERS:
+ // ...
+ break;
+ case XMLStreamConstants.END_ELEMENT:
+ // ...
+ endLocalName = reader.getLocalName();
+ if ("Object".equalsIgnoreCase(endLocalName)) {
+ objectFlag = false;
+ pmDatas.putAll(commonNameAndValue);
+ try {
+ pmResultChannel.put(pmDatas);
+
+ } catch (InterruptedException e) {
+ log.error("collectResultChannel.put(resultMap) error ", e);
+ }
+ // System.out.println(pmDatas);
+ // pmDatas.clear();
+ }
+ if (endLocalName.equalsIgnoreCase("PmData")) {
+ pmDataFlag = false;
+ }
+
+ if (endLocalName.equalsIgnoreCase("PmName")) {
+ pmNameFlag = false;
+ }
+ if (endLocalName.equalsIgnoreCase("Measurements")) {
+ // a measurement over
+ measurementStart = false;
+ }
+
+ if ("FileHeader".equalsIgnoreCase(endLocalName)) {
+ fileHeaderStart = false;
+ }
+ break;
+ }
+ } catch (Exception e) {
+ log.error("", e);
+ event = reader.next();
+ }
+ }
+
+ } catch (Exception e) {
+ log.error("processPMXml is Exception ", e);
+ return false;
+ } finally {
+ try {
+ if (reader != null)
+ reader.close();
+ if (isr != null)
+ isr.close();
+ if (fis != null)
+ fis.close();
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+ return true;
+ }
+
+ private String getXMLAttribute(XMLStreamReader reader, String obj) {
+ String res = null;
+ if (obj == null || reader == null) {
+ return res;
+ }
+ int n = reader.getAttributeCount();
+ for (int i = 0; i < n; i++) {
+ String name = reader.getAttributeLocalName(i);
+ if (obj.equalsIgnoreCase(name)) {
+ res = reader.getAttributeValue(i);
+ }
+ }
+ return res;
+ }
+
+ public boolean processPMCsv(File tempfile) {
+
+ FileInputStream brs = null;
+ InputStreamReader isr = null;
+ BufferedReader br = null;
+
+ List<String> columnNames = new ArrayList<String>();
+ List<String> commonValues = new ArrayList<String>();
+ try {
+
+ brs = new FileInputStream(tempfile);
+ isr = new InputStreamReader(brs, Constant.ENCODING_UTF8);
+ br = new BufferedReader(isr);
+ // common field
+ String commonField = br.readLine();
+ String[] fields = commonField.split("\\|", -1);
+ for (String com : fields) {
+ String[] comNameAndValue = com.split("=", 2);
+ columnNames.add(comNameAndValue[0].trim());
+ commonValues.add(comNameAndValue[1]);
+ }
+ // column names
+ String columnName = br.readLine();
+ String[] names = columnName.split("\\|", -1);
+ for (String name : names) {
+ columnNames.add(name);
+ }
+
+ String valueLine = "";
+ List<String> valuelist = new ArrayList<String>();
+
+ while ((valueLine = br.readLine()) != null) {
+ if (valueLine.trim().equals("")) {
+ continue;
+ }
+ // countNum ++;
+ String[] values = valueLine.split("\\|", -1);
+
+ valuelist.addAll(commonValues);
+ for (String value : values) {
+ valuelist.add(value);
+ }
+ // this.appendLine(valuelist, bos);
+ // resultMap
+ HashMap<String, String> resultMap = this.resultMap(columnNames, valuelist);
+ try {
+ pmResultChannel.put(resultMap);
+ } catch (InterruptedException e) {
+ log.error("collectResultChannel.put(resultMap) error ", e);
+ throw new RuntimeException(e);
+ }
+ valuelist.clear();
+ }
+ } catch (IOException e) {
+ log.error("processPMCsv is fail ", e);
+ return false;
+ } finally {
+ try {
+ if (br != null)
+ br.close();
+ if (isr != null)
+ isr.close();
+ if (brs != null)
+ brs.close();
+
+ } catch (Exception e) {
+ log.error(e);
+ }
+ }
+ return true;
+
+ }
+
+ private HashMap<String, String> resultMap(List<String> columnNames, List<String> valuelist) {
+
+ HashMap<String, String> resultMap = new HashMap<String, String>();
+ if (columnNames.size() == valuelist.size()) {
+ for (int i = 0; i < columnNames.size(); i++) {
+ resultMap.put(columnNames.get(i), valuelist.get(i));
+ }
+ }
+
+ return resultMap;
+
+ }
+
+ private boolean processCMXml(File tempfile, String nename, String type) {
+
+ String csvpath = localPath + nename + "/" + type + "/";
+ File csvpathfile = new File(csvpath);
+ if (!csvpathfile.exists()) {
+ csvpathfile.mkdirs();
+ }
+ String csvFileName = nename + dateFormat.format(new Date()) + System.nanoTime();
+ String csvpathAndFileName = csvpath + csvFileName + ".csv";
+ BufferedOutputStream bos = null;
+ FileOutputStream fos = null;
+ try {
+ fos = new FileOutputStream(csvpathAndFileName, false);
+ bos = new BufferedOutputStream(fos, 10240);
+ } catch (FileNotFoundException e1) {
+ log.error("FileNotFoundException " + StringUtil.getStackTrace(e1));
+ }
+
+ boolean FieldNameFlag = false;
+ boolean FieldValueFlag = false;
+ // line num
+ int countNum = 0;
+ String xmlPathAndFileName = null;
+ String localName = null;
+ String endLocalName = null;
+ String rmUID = null;
+ int index = -1;
+ ArrayList<String> names = new ArrayList<String>();// colname
+ LinkedHashMap<String, String> nameAndValue = new LinkedHashMap<String, String>();
+
+ FileInputStream fis = null;
+ InputStreamReader isr = null;
+ XMLStreamReader reader = null;
+ try {
+ fis = new FileInputStream(tempfile);
+ isr = new InputStreamReader(fis, Constant.ENCODING_UTF8);
+ XMLInputFactory fac = XMLInputFactory.newInstance();
+ reader = fac.createXMLStreamReader(isr);
+ int event = -1;
+ boolean setcolum = true;
+ while (reader.hasNext()) {
+ try {
+ event = reader.next();
+ switch (event) {
+ case XMLStreamConstants.START_ELEMENT:
+ localName = reader.getLocalName();
+ if ("FieldName".equalsIgnoreCase(localName)) {
+ FieldNameFlag = true;
+ }
+ if (FieldNameFlag) {
+ if ("N".equalsIgnoreCase(localName)) {
+ String colName = reader.getElementText().trim();
+ names.add(colName);
+ }
+ }
+ if ("FieldValue".equalsIgnoreCase(localName)) {
+ FieldValueFlag = true;
+
+ }
+ if (FieldValueFlag) {
+ if (setcolum) {
+ xmlPathAndFileName = this.setColumnNames(nename, names, type);
+ setcolum = false;
+ }
+
+ if ("Object".equalsIgnoreCase(localName)) {
+ int ac = reader.getAttributeCount();
+ for (int i = 0; i < ac; i++) {
+ if ("rmUID".equalsIgnoreCase(reader.getAttributeLocalName(i))) {
+ rmUID = reader.getAttributeValue(i).trim();
+ }
+ }
+ nameAndValue.put("rmUID", rmUID);
+ }
+ if ("V".equalsIgnoreCase(localName)) {
+ index = Integer.parseInt(reader.getAttributeValue(0)) - 1;
+ String currentName = names.get(index);
+ String v = reader.getElementText().trim();
+ nameAndValue.put(currentName, v);
+ }
+ }
+ break;
+ case XMLStreamConstants.CHARACTERS:
+ break;
+ case XMLStreamConstants.END_ELEMENT:
+ endLocalName = reader.getLocalName();
+
+ if ("FieldName".equalsIgnoreCase(endLocalName)) {
+ FieldNameFlag = false;
+ }
+ if ("FieldValue".equalsIgnoreCase(endLocalName)) {
+ FieldValueFlag = false;
+ }
+ if ("Object".equalsIgnoreCase(endLocalName)) {
+ countNum++;
+ this.appendLine(nameAndValue, bos);
+ nameAndValue.clear();
+ }
+ break;
+ }
+ } catch (Exception e) {
+ log.error("" + StringUtil.getStackTrace(e));
+ event = reader.next();
+ }
+ }
+
+ if (bos != null) {
+ bos.close();
+ bos = null;
+ }
+ if (fos != null) {
+ fos.close();
+ fos = null;
+ }
+
+ String[] fileKeys = this.createZipFile(csvpathAndFileName, xmlPathAndFileName, nename);
+ // ftp store
+ Properties ftpPro = configurationInterface.getProperties();
+ String ip = ftpPro.getProperty("ftp_ip");
+ String port = ftpPro.getProperty("ftp_port");
+ String ftp_user = ftpPro.getProperty("ftp_user");
+ String ftp_password = ftpPro.getProperty("ftp_password");
+
+ String ftp_passive = ftpPro.getProperty("ftp_passive");
+ String ftp_type = ftpPro.getProperty("ftp_type");
+ String remoteFile = ftpPro.getProperty("ftp_remote_path");
+ this.ftpStore(fileKeys, ip, port, ftp_user, ftp_password, ftp_passive, ftp_type, remoteFile);
+ // create Message
+ String message = this.createMessage(fileKeys[1], ftp_user, ftp_password, ip, port, countNum, nename);
+
+ // set message
+ this.setMessage(message);
+ } catch (Exception e) {
+ log.error("" + StringUtil.getStackTrace(e));
+ return false;
+ } finally {
+ try {
+ if (reader != null) {
+ reader.close();
+ }
+ if (isr != null) {
+ isr.close();
+ }
+ if (fis != null) {
+ fis.close();
+ }
+ if (bos != null) {
+ bos.close();
+ }
+
+ if (fos != null) {
+ fos.close();
+ }
+ } catch (Exception e) {
+ log.error(e);
+ }
+ }
+ return true;
+ }
+
+ private void setMessage(String message) {
+
+ try {
+ cmResultChannel.put(message);
+ } catch (Exception e) {
+ log.error("collectResultChannel.put(message) is error " + StringUtil.getStackTrace(e));
+ }
+ }
+
+ public String createMessage(String zipName, String user, String pwd, String ip, String port, int countNum,
+ String nename) {
+
+ StringBuffer strBuffer = new StringBuffer();
+ strBuffer.append("<?xml version=\"1.0\" encoding=\"UTF-8\"?>"
+ + "<FILE_DATA_READY_UL xmlns:xsi=\" http://www.w3.org/2001/XMLSchema-instance\">"
+ + "<Header SessionID=\"");
+ strBuffer.append("");
+ strBuffer.append("\" LicenceID=\"");
+ strBuffer.append("");
+ strBuffer.append("\" SystemID=\"");
+ strBuffer.append("");
+ strBuffer.append("\" Time=\"");
+ strBuffer.append(dateFormat2.format(new Date()));
+ strBuffer.append("\" PolicyID=\"");
+ strBuffer.append("");
+ strBuffer.append("\"/><Body>");
+ strBuffer.append("<DataCatalog>");
+ strBuffer.append("");
+ strBuffer.append("</DataCatalog><GroupID>");
+ strBuffer.append(nename);
+ strBuffer.append("</GroupID><DataSourceName>");
+ strBuffer.append("");
+ strBuffer.append("</DataSourceName><InstanceID>");
+ strBuffer.append("");
+ strBuffer.append("</InstanceID><FileFormat>");
+ strBuffer.append("csv");
+ strBuffer.append("</FileFormat><CharSet>");
+ strBuffer.append("gbk");
+ strBuffer.append("</CharSet><FieldSeparator>");
+ strBuffer.append("|");
+ strBuffer.append("</FieldSeparator><IsCompressed>");
+ strBuffer.append("true");
+ strBuffer.append("</IsCompressed><StartTime>");
+ strBuffer.append(dateFormat2.format(new Date()));
+ strBuffer.append("</StartTime><EndTime>");
+ strBuffer.append("");
+ strBuffer.append("</EndTime><FileList>");
+ strBuffer.append(zipName);
+ strBuffer.append("</FileList><ConnectionString>");
+ strBuffer.append("ftp://" + user + ":" + pwd + "@" + ip + ":" + port);
+ strBuffer.append("</ConnectionString>");
+ strBuffer.append("<DataCount>");
+ strBuffer.append(countNum);
+ strBuffer.append("</DataCount>");
+
+ strBuffer.append("<FileSize>").append("").append("</FileSize>");
+ strBuffer.append("<DataGranularity>").append("").append("</DataGranularity>");
+
+ strBuffer.append("</Body></FILE_DATA_READY_UL>");
+ return strBuffer.toString();
+
+ }
+
+ private void ftpStore(String[] fileKeys, String ip, String port, String ftp_user, String ftp_password,
+ String ftp_passive, String ftp_type, String remoteFile) {
+ String zipFilePath = fileKeys[0];
+
+ FTPInterface ftpClient;
+ ftpClient = new FTPSrv();
+ // login
+ try {
+ ftpClient.login(ip, Integer.parseInt(port), ftp_user, ftp_password, "GBK",
+ Boolean.parseBoolean(ftp_passive), 5 * 60 * 1000);
+ } catch (Exception e) {
+ log.error("login fail,ip=[" + ip + "] port=[" + port + "] user=[" + ftp_user + "]pwd=[" + ftp_password + "]"
+ + StringUtil.getStackTrace(e));
+ return;
+ }
+ // ftpClient.store(zipFilePath, remoteFile);
+ log.debug("store [" + zipFilePath + "]to[" + remoteFile + "]");
+
+ FileUtils.deleteQuietly(new File(zipFilePath));
+
+ }
+
+ private String[] createZipFile(String csvpathAndFileName, String xmlPathAndFileName, String nename)
+ throws IOException {
+
+ String zipPath = resultPath + nename + dateFormat.format(new Date()) + "_" + System.nanoTime();
+
+ File destDir = new File(zipPath);
+ destDir.mkdirs();
+
+ try {
+ FileUtils.copyFileToDirectory(new File(csvpathAndFileName), destDir);
+ FileUtils.copyFileToDirectory(new File(xmlPathAndFileName), destDir);
+ } catch (IOException e) {
+ throw e;
+ // flow should end here in case of exception
+ }
+
+ String destFilePath = zipPath + ".zip";
+ try {
+ Zip zip = new Zip(destDir.getAbsolutePath(), destFilePath);
+ zip.setCompressLevel(9);
+ zip.compress();
+
+ FileUtils.deleteDirectory(destDir);
+ } catch (IOException e) {
+ log.error("zip.compress() is fail " + StringUtil.getStackTrace(e));
+ }
+ return new String[] { destFilePath, zipPath + ".zip" };
+ }
+
+ private String setColumnNames(String nename, List<String> names, String type) {
+ // write xml
+ String xmlpath = localPath + nename + "/" + type + "/";
+ File xmlpathfile = new File(xmlpath);
+ if (!xmlpathfile.exists()) {
+ xmlpathfile.mkdirs();
+ }
+ String xmlFileName = nename + dateFormat.format(new Date()) + System.nanoTime();
+ String fieldLine = "";
+ for (int i = 0; i < names.size(); i++) {
+ String field = "\t<Field>\r\n" + "\t\t<FieldNo>" + i + "</FieldNo>\r\n" + "\t\t<FieldName>" + names.get(i)
+ + "</FieldName>\r\n" + "\t\t<FieldType>" + names.get(i) + "</FieldType>\r\n"
+ + "\t\t<FieldNameOther>" + names.get(i) + "</FieldNameOther>\r\n" + "\t</Field>\r\n";
+ fieldLine = fieldLine + field;
+ }
+
+ String str = "<?xml version=\"1.0\" encoding=\"gbk\"?>\r\n" + "<xml>\r\n" + "<FILE_STRUCTURE>\r\n" + fieldLine
+ + "</FILE_STRUCTURE>\r\n" + "</xml>\r\n";
+ String xmlPathAndFileName = xmlpath + xmlFileName + ".xml";
+ try {
+ this.writeDetail(xmlPathAndFileName, str);
+ } catch (Exception e) {
+ log.error("writeDetail is fail ,xmlFileName=" + xmlFileName + StringUtil.getStackTrace(e));
+ }
+
+ return xmlPathAndFileName;
+ }
+
+ private void writeDetail(String detailFileName, String str) throws Exception {
+
+ try (OutputStream readOut = new FileOutputStream(new File(detailFileName), false);
+ OutputStreamWriter writer = new OutputStreamWriter(readOut)) {
+ writer.write(str);
+ writer.flush();
+ } catch (IOException e) {
+ throw e;
+ }
+ }
+
+ private void appendLine(LinkedHashMap<String, String> nameAndValue, BufferedOutputStream bos) {
+ StringBuilder lineDatas = new StringBuilder();
+
+ for (String key : nameAndValue.keySet()) {
+ lineDatas.append(nameAndValue.get(key)).append("|");
+ }
+ try {
+ bos.write(lineDatas.toString().getBytes());
+ bos.write("\n".getBytes());
+ } catch (IOException e) {
+ log.error("appendLine error " + StringUtil.getStackTrace(e));
+ }
+ }
+
+ // private void appendLine(List<String> values,BufferedOutputStream bos) {
+ // StringBuilder lineDatas = new StringBuilder();
+ //
+ // for (String value : values) {
+ // lineDatas.append(value).append("|");
+ // }
+ // try {
+ // bos.write(lineDatas.toString().getBytes());
+ // bos.write("\n".getBytes());
+ // } catch (IOException e) {
+ // log.error("appendLine error "+StringUtil.getStackTrace(e));
+ // }
+ // }
+
+ public List<File> decompressed(String fileName) {
+ List<File> filelist = new ArrayList<File>();
+
+ if (fileName.indexOf(".gz") > 1) {
+ try {
+ File decompressFile = deGz(fileName);
+ filelist.add(decompressFile);
+ new File(fileName).delete();
+ } catch (IOException e) {
+ log.error("decompressed is fail " + StringUtil.getStackTrace(e));
+ }
+ } else if (fileName.indexOf(".zip") > 1) {
+ try {
+ File[] files = deZip(new File(fileName));
+ new File(fileName).delete();
+ for (File temp : files) {
+ filelist.add(temp);
+ }
+ } catch (Exception e) {
+ log.error("decompressed is fail " + StringUtil.getStackTrace(e));
+ }
+ } else {
+ filelist.add(new File(fileName));
+ }
+
+ return filelist;
+ }
+
+ private File deGz(String gzFileName) throws IOException {
+ Gunzip gunzip = new Gunzip();
+ String orgFile = gzFileName.replace(".gz", "");
+ gunzip.unCompress(gzFileName, orgFile);
+ return new File(orgFile);
+ }
+
+ public File[] deZip(File file) throws Exception {
+
+ String regx = "(.*).zip";
+ Pattern p = Pattern.compile(regx);
+ Matcher m = p.matcher(file.getName());
+ if (m.find()) {
+ String orgFile = localPath + m.group(1) + "/";
+ UnZip unzip = new UnZip(file.getAbsolutePath(), orgFile);
+ unzip.deCompress();
+ file = new File(orgFile);
+ }
+ File[] files = file.listFiles();
+
+ return files;
+
+ }
+
+ private List<String> ftpDownload(CollectVo collectVo) {
+
+ List<String> fileList = new ArrayList<String>();
+ // IP
+ String ip = collectVo.getIP();
+ // port
+ String port = collectVo.getPort();
+ // user
+ String user = collectVo.getUser();
+ // password
+ String password = collectVo.getPassword();
+ // isPassiveMode
+ String passivemode = collectVo.getPassive();
+
+ FTPInterface ftpClient = new FTPSrv();
+
+ // login
+ try {
+ log.info("ftp login ,ip=[" + ip + "] port=[" + port + "] user=[" + user + "]password=[" + password + "]");
+ ftpClient.login(ip, Integer.parseInt(port), user, password, "GBK", Boolean.parseBoolean(passivemode),
+ 5 * 60 * 1000);
+ } catch (Exception e) {
+ log.error("login fail,ip=[" + ip + "] port=[" + port + "] user=[" + user + "]password=[" + password + "]"
+ + StringUtil.getStackTrace(e));
+ return fileList;
+ }
+
+ // download
+ String dir = collectVo.getRemotepath();
+ List<String> searchExprList = new ArrayList<String>();
+ String[] FPath = dir.split(";");
+ for (int i = 0; i < FPath.length; i++) {
+ int oldSize = searchExprList.size();
+ String conpath = FPath[i] + collectVo.getMatch();
+ Hashtable<String, String> varMap = new Hashtable<String, String>();
+ long collectPeriod = 900;
+ try {
+ collectPeriod = Long.parseLong(collectVo.getGranularity()) * 60;
+ log.info("collectPeriod =[" + collectPeriod + "]");
+ } catch (NumberFormatException e) {
+ e.printStackTrace();
+ }
+ long[] d = DateUtil.getScanScope(new Date(), collectPeriod);
+ searchExprList.add(VarExprParser.replaceVar(conpath, d[0], d[1]));
+
+ varMap.clear();
+ varMap = null;
+ log.info("[" + conpath + "] result[" + (searchExprList.size() - oldSize) + "] path");
+ conpath = null;
+ }
+ String nowdir = null;
+ try {
+ nowdir = ftpClient.pwd();
+ searchExprList = getPathNoRegular(searchExprList, ftpClient);
+ } catch (Exception e1) {
+ log.error(" collect fail ", e1);
+ return fileList;
+ }
+ List<AFtpRemoteFile> remoteFiles = new ArrayList<AFtpRemoteFile>();
+ for (String expr : searchExprList) {
+ ftpClient.chdir(nowdir);
+ String keys[] = parseExprKeys(expr);
+ String ftpRegular = keys[1];
+ String ftpDir = keys[0];
+
+ boolean cdsucess = ftpClient.chdir(ftpDir);
+ if (cdsucess) {
+ AFtpRemoteFile[] arf = (AFtpRemoteFile[]) ftpClient.list();
+ log.info(" list [" + ftpDir + "] result[" + (arf == null ? "null" : arf.length) + "] files");
+ // filter
+
+ rfileFilter(remoteFiles, arf, ftpRegular);
+
+ keys = null;
+ ftpRegular = ftpDir = null;
+
+ for (AFtpRemoteFile ftpRemoteFile : remoteFiles) {
+ if (!new File(localPath).exists()) {
+ try {
+ new File(localPath).mkdir();
+ } catch (Exception e) {
+ log.error("create localPath is fail localPath=" + localPath + " "
+ + StringUtil.getStackTrace(e));
+ }
+ }
+
+ if (!new File(localPath).exists()) {
+ new File(localPath).mkdirs();
+ }
+
+ String localFileName = localPath + ftpRemoteFile.getFileName();
+ File loaclFile = new File(localFileName);
+ if (loaclFile.exists()) {
+ loaclFile.delete();
+ }
+
+ boolean flag = ftpClient.downloadFile(ftpRemoteFile.getAbsFileName(), localFileName);
+
+ if (flag) {
+ fileList.add(localFileName);
+ } else {
+ log.error("download file fail fileName=" + ftpRemoteFile.getAbsFileName());
+ }
+ }
+
+ } else {
+ log.error("cd dir is faill dir =[" + ftpDir + "]");
+ }
+ }
+
+ return fileList;
+ }
+
+ private void rfileFilter(List<AFtpRemoteFile> fileContainer, AFtpRemoteFile[] arfs, String ftpRegular) {
+ if (ftpRegular != null && ftpRegular.length() > 0) {
+ Pattern pattern = null;
+ try {
+ pattern = Pattern.compile(ftpRegular, Pattern.CASE_INSENSITIVE);
+ } catch (Exception e) {
+ log.info("[" + ftpRegular + "]Pattern.compile exception:" + e.getMessage());
+ // should rethrow exception or return from here
+ }
+ int hisSize = fileContainer.size();
+ for (int j = 0; arfs != null && j < arfs.length; j++) {
+ String fileName = parseFileName(arfs[j].getFileName());
+ Matcher matcher = null;
+ if (pattern != null)
+ matcher = pattern.matcher(fileName);
+ else {
+ // define the flow when pattern is null
+ }
+
+ if (matcher.find())
+ fileContainer.add(arfs[j]);
+ }
+ log.info("[" + ftpRegular + "]filter[" + (fileContainer.size() - hisSize) + "]filse");
+ pattern = null;
+ } else {
+ for (int j = 0; arfs != null && j < arfs.length; j++)
+ fileContainer.add(arfs[j]);
+ }
+
+ }
+
+ private String parseFileName(String fileName) {
+ int idx = fileName.lastIndexOf("/");
+ if (idx == -1)
+ return fileName;
+ return fileName.substring(idx + 1, fileName.length());
+ }
+
+ private String[] parseExprKeys(String source) {
+
+ if (source.indexOf(";") > -1) {
+ source = source.substring(0, source.indexOf(";"));
+ }
+ if (source.endsWith("/"))
+ return new String[] { source, "" };
+
+ int idx = source.lastIndexOf("/");
+ String[] dirkeys = new String[2];
+ dirkeys[0] = source.substring(0, idx + 1);
+ dirkeys[1] = source.substring(idx + 1, source.length());
+ return dirkeys;
+ }
+
+ public List<String> getPathNoRegular(List<String> searchExprList, FTPInterface ftpCache) throws Exception {
+ boolean isregular = false;
+ List<String> regularList = new ArrayList<String>();
+ for (String regular : searchExprList) {
+ Pattern lpattern = null;
+ try {
+ lpattern = Pattern.compile("(.*/)<([^/]+)>(/.*)");
+ } catch (Exception e) {
+ log.error("[" + regular + "]compile fails:" + e.getMessage());
+ e.printStackTrace();
+ }
+ Matcher matcher = null;
+ if (lpattern != null)
+ matcher = lpattern.matcher(regular);
+ else {
+ // define flow in case lpattern is null
+ }
+
+ if (matcher.find()) {
+ isregular = true;
+ String parpath = matcher.group(1);
+ try {
+ boolean isin = ftpCache.chdir(parpath);
+ if (isin) {
+ log.info("cd dir [" + parpath + "] sucess");
+ } else {
+ log.error("cd dir [" + parpath + "] fail");
+ }
+ } catch (Exception e) {
+ log.error(" cd dir [" + parpath + "]fail", e);
+ throw e;
+ }
+ RemoteFile[] remotef = ftpCache.list();
+ for (RemoteFile aremote : remotef) {
+ if (aremote.isDirectory() && aremote.getFileName().matches(matcher.group(2))) {
+ regularList.add(matcher.group(1) + aremote.getFileName() + matcher.group(3));
+ }
+ }
+ } else {
+ regularList.add(regular);
+ }
+ }
+ if (isregular == true) {
+ getPathNoRegular(regularList, ftpCache);
+ }
+ return regularList;
+ }
}
Code Review / vfc / nfvo / driver / ems.git / commitdiff