1 package user.jobengine.server.steps;
\r
3 import java.io.OutputStream;
\r
4 import java.net.InetAddress;
\r
6 import java.text.ParseException;
\r
7 import java.text.SimpleDateFormat;
\r
8 import java.util.Calendar;
\r
9 import java.util.Date;
\r
10 import java.util.List;
\r
12 import org.apache.commons.lang.StringUtils;
\r
13 import org.apache.commons.net.ftp.FTP;
\r
14 import org.apache.commons.net.ftp.FTPClient;
\r
15 import org.apache.commons.net.ftp.FTPReply;
\r
16 import org.apache.logging.log4j.LogManager;
\r
17 import org.apache.logging.log4j.Logger;
\r
18 import org.apache.logging.log4j.Marker;
\r
20 import com.ibm.nosql.json.api.BasicDBList;
\r
21 import com.ibm.nosql.json.api.BasicDBObject;
\r
22 import com.ibm.nosql.json.api.DB;
\r
23 import com.ibm.nosql.json.api.DBCollection;
\r
24 import com.ibm.nosql.json.api.DBCursor;
\r
25 import com.ibm.nosql.json.api.DBObject;
\r
26 import com.ibm.nosql.json.api.QueryBuilder;
\r
28 import user.commons.CalendarUtils;
\r
29 import user.commons.ListUtils;
\r
30 import user.commons.StoreUri;
\r
31 import user.commons.nosql.NoSQLUtils;
\r
32 import user.commons.octopus.IOctopusAPI;
\r
33 import user.commons.octopus.OctopusAPI;
\r
34 import user.commons.remotestore.FtpDirectoryLister;
\r
35 import user.commons.remotestore.RemoteStoreProtocol;
\r
36 import user.jobengine.db.IItemManager;
\r
37 import user.jobengine.server.IJobEngine;
\r
38 import user.jobengine.server.IJobRuntime;
\r
40 public class CopyForArchiveNEXIORecordingsStep extends JobStep {
\r
41 private static final String MEDIATYPE = "Visszarögzített";
\r
42 private static final String SCHEDULED_FORMAT = "yyyy.MM.dd HH:mm";
\r
43 private static final String STARTTIME_FORMAT = "HHmm";
\r
44 private static final String RUNDOWNDATE_FORMAT = "yyyyMMdd";
\r
45 private static final Logger logger = LogManager.getLogger();
\r
46 private static final String UTF_8 = "utf-8";
\r
47 private static final String JSON_EXT = ".json";
\r
48 private static final String XML_EXT = ".xml";
\r
49 private static final String DURATION = "duration";
\r
50 private static final String MXFEXT = ".MXF";
\r
51 private static final String NEXIOCLIPS = "nexioclips";
\r
52 private static final String LONGNAMEID = "longnameid";
\r
53 private static final String EXTAGENCY = "extagency";
\r
54 private static final String RECORDDATE = "recorddate";
\r
55 private static final SimpleDateFormat startTimeformat = new SimpleDateFormat(STARTTIME_FORMAT);
\r
56 private static final SimpleDateFormat rundownDateformat = new SimpleDateFormat(RUNDOWNDATE_FORMAT);
\r
58 private OctopusAPI octopusAPI;
\r
59 private IItemManager manager;
\r
62 private FTPClient sourceFtp;
\r
63 private FTPClient targetFtp;
\r
64 private StoreUri sourceUri;
\r
65 private StoreUri targetUri;
\r
66 private int nexioKillDateDays;
\r
67 private String nexioAgency;
\r
68 private Object[] filterAgencies;
\r
69 private Marker systemMarker;
\r
71 private int check(int value, String name) {
\r
73 logger.error(systemMarker, "A folyamat '{}' bemeneti paramétere 0.", name);
\r
74 throw new NullPointerException(String.format("System is not configured properly, missing '%s' input parameter.", name));
\r
79 private String check(String value, String name) {
\r
80 if (StringUtils.isBlank(value)) {
\r
81 logger.error(systemMarker, "A folyamat '{}' bemeneti paramétere üres.", name);
\r
82 throw new NullPointerException(String.format("System is not configured properly, missing '%s' input parameter.", name));
\r
87 private void copy(RundownArchive rundownArchive) {
\r
88 for (StoryArchive storyArchive : rundownArchive.getStoryArchives()) {
\r
89 for (FileArchive fileArchive : storyArchive.getFileArchives()) {
\r
91 copyFile(fileArchive, rundownArchive, storyArchive);
\r
92 logger.info(systemMarker, "Sikeres fájl archiválás a '{}' tükörhöz: '{}'.", rundownArchive.getItemTitle(), fileArchive.getFileName());
\r
93 } catch (Exception e) {
\r
94 logger.error(systemMarker, "Az '{}' fájl archiválása sikertelen. A rendszer üzenete: {}", fileArchive.getFileName(), e.getMessage());
\r
100 private void copyFile(FileArchive fileArchive, RundownArchive rundownArchive, StoryArchive storyArchive) throws Exception {
\r
101 String sourceFileName = fileArchive.getFileName();
\r
102 String targetFileName = getTargetFileName(rundownArchive, sourceFileName);
\r
103 transferFile(sourceFileName, targetFileName);
\r
104 BasicDBObject metadata = createMetadata(rundownArchive, storyArchive, fileArchive);
\r
105 transferMetadata(targetFileName, metadata);
\r
106 createSourceKillDateFile(rundownArchive, sourceFileName);
\r
109 private BasicDBObject createMetadata(RundownArchive rundownArchive, StoryArchive storyArchive, FileArchive fileArchive) {
\r
110 BasicDBObject result = new BasicDBObject();
\r
111 result.put("itemHouseId", rundownArchive.getItemHouseId());
\r
112 result.put("itemTitle", rundownArchive.getItemTitle());
\r
113 result.put("itemDescription", rundownArchive.getItemDesc());
\r
114 result.put("userName", "mediacube");
\r
116 result.put("mediaHouseId", storyArchive.getMediaHouseId());
\r
117 result.put("mediaTitle", storyArchive.getMediaTitle());
\r
118 result.put("mediaDescription", storyArchive.getMediaDesc());
\r
120 result.put("duration", fileArchive.getDuration());
\r
121 result.put("mediaType", MEDIATYPE);
\r
125 private void createSourceKillDateFile(RundownArchive rundownArchive, String fileName) throws Exception {
\r
126 logger.info("Create killdate/agency for {}", fileName);
\r
127 OutputStream outStream = null;
\r
129 sourceFtp = ((FtpDirectoryLister) sourceUri.getLister()).connect();
\r
130 Calendar killDate = CalendarUtils.createCalendar(rundownArchive.getScheduleDate());
\r
131 killDate.add(Calendar.DAY_OF_YEAR, nexioKillDateDays);
\r
132 byte[] killDateFile = EscortFiles.createNEXIOKillDateFile(fileName, killDate.getTime(), null, nexioAgency);
\r
133 outStream = sourceFtp.storeFileStream(fileName + XML_EXT);
\r
134 if (outStream == null) {
\r
135 throw new NullPointerException("Can not open: " + fileName + XML_EXT + " Reply:" + sourceFtp.getReplyString());
\r
137 outStream.write(killDateFile);
\r
139 } catch (Exception e) {
\r
140 logger.catching(e);
\r
143 if (outStream != null)
\r
145 sourceUri.cleanUp();
\r
150 public Object[] execute(int nexioPort, String nexioUserName, String nexioPassword, String archiveFtp, String archiveUserName, String archivePassword,
\r
151 String filterAgencies, int limit, int nexioKillDateDays, String targetAgency, IJobEngine jobEngine, IJobRuntime jobRuntime) throws Exception {
\r
152 systemMarker = jobRuntime.getMarker();
\r
153 setAndCheck(nexioPort, nexioUserName, nexioPassword, archiveFtp, archiveUserName, archivePassword, filterAgencies, limit, nexioKillDateDays,
\r
154 targetAgency, jobEngine);
\r
155 octopusAPI = new OctopusAPI();
\r
156 List<BasicDBObject> clips = queryClips();
\r
157 processClips(clips, limit);
\r
161 private Date getScheduledStart(String clipName, Date recordDate) {
\r
162 if (StringUtils.isBlank(clipName)) {
\r
163 logger.warn(systemMarker, "A fájlnak nincs neve, ezért nem archiválható.");
\r
166 if (recordDate == null) {
\r
167 logger.warn(systemMarker, "Az '{}' fájl rögzítésének ideje nem meghatározható, ezért nem archiválható.", clipName);
\r
171 Date timePart = null;
\r
173 String clipNameTime = clipName.split("_")[0];
\r
174 timePart = startTimeformat.parse(clipNameTime);
\r
175 } catch (ParseException e) {
\r
176 logger.warn(systemMarker, "A '{}' fájl neve nem időbélyeggel kezdődik, ezért nem archiválható.", clipName);
\r
179 return CalendarUtils.createCalendar(CalendarUtils.createCalendar(recordDate), timePart).getTime();
\r
182 private String getTargetFileName(RundownArchive rundownArchive, String sourceFileName) {
\r
183 String date = rundownDateformat.format(rundownArchive.getScheduleDate());
\r
184 return String.format("%s-%s%s", date, sourceFileName, MXFEXT);
\r
187 private RundownArchive processClip(BasicDBObject clip) {
\r
188 RundownArchive result = null;
\r
189 String clipName = NoSQLUtils.asString(clip, LONGNAMEID);
\r
190 Date recordDate = clip.getDate(RECORDDATE);
\r
191 long duration = NoSQLUtils.asLong(clip, DURATION);
\r
192 logger.info("Processing {} {}", clipName, recordDate);
\r
193 Date scheduledStart = getScheduledStart(clipName, recordDate);
\r
194 if (scheduledStart == null)
\r
196 DBObject rundown = octopusAPI.getRundown(scheduledStart);
\r
197 if (rundown == null) {
\r
198 logger.error(systemMarker, "A '{}' anyaghoz nem található tükör '{}' kezdéssel, ezért nem archiválható.", clipName, scheduledStart);
\r
203 result = processRundow(rundown, clipName, duration);
\r
204 } catch (Exception e) {
\r
205 logger.catching(e);
\r
206 logger.error(systemMarker, "A '{}' anyag metaadatainak transzformálása sikertelen, ezért az nem archiválható. A rendszer hibaüzenete: {}",
\r
211 if (clipName.startsWith("1900")) {
\r
212 Calendar cal = CalendarUtils.createCalendar(scheduledStart);
\r
213 cal.add(Calendar.MINUTE, 5);
\r
214 rundown = octopusAPI.getRundown(cal.getTime());
\r
215 if (rundown == null) {
\r
216 logger.error(systemMarker, "A '{}' anyaghoz nem található tükör '{}' kezdéssel, ezért nem archiválható.", clipName, scheduledStart);
\r
220 RundownArchive item2 = null;
\r
223 item2 = processRundow(rundown, clipName, duration);
\r
224 } catch (Exception e) {
\r
225 logger.catching(e);
\r
226 logger.error(systemMarker, "A '{}' anyag metaadatainak transzformálása sikertelen, ezért az nem archiválható. A rendszer hibaüzenete: {}",
\r
231 result.setItemTitle(result.getItemTitle() + " + NAPIAKT");
\r
232 StoryArchive storyArchive = result.getStoryArchives().get(0);
\r
233 StoryArchive storyArchive2 = item2.getStoryArchives().get(0);
\r
234 storyArchive.setMediaDesc(storyArchive.getMediaDesc() + "\r\n\r\n****** NAPIAKT ******\r\n\r\n" + storyArchive2.getMediaDesc());
\r
240 private void processClips(List<BasicDBObject> clips, int limit) {
\r
241 logger.info(systemMarker, "A folyamat {} archiválható anyagot érzékelt.", clips.size());
\r
243 for (BasicDBObject clip : clips) {
\r
244 RundownArchive rundownArchive = processClip(clip);
\r
245 if (rundownArchive == null)
\r
248 copy(rundownArchive);
\r
251 if (current == limit)
\r
253 setProgress(current * 100 / limit);
\r
257 private RundownArchive processRundow(DBObject r, String clipName, long duration) throws Exception {
\r
258 BasicDBObject rundown = (BasicDBObject) r;
\r
259 long rundownID = rundown.getLong(IOctopusAPI.ID);
\r
260 logger.info("Processing rundown {} {}", rundownID, rundown.getString(IOctopusAPI.NAME));
\r
262 List<DBObject> stories = octopusAPI.getRundownFullStories(rundownID);
\r
263 if (stories == null)
\r
265 RundownArchive result = new RundownArchive();
\r
267 String name = NoSQLUtils.asString(NoSQLUtils.asDBObject(rundown, IOctopusAPI.RUNDOWN_TYPE), IOctopusAPI.NAME);
\r
268 if (StringUtils.isBlank(name))
\r
270 String channel = NoSQLUtils.asString(NoSQLUtils.asDBObject(rundown, IOctopusAPI.CHANNEL), IOctopusAPI.NAME);
\r
271 Date scheduledStart = rundown.getDate(IOctopusAPI.SCHEDULED_START);
\r
272 if (scheduledStart == null)
\r
274 String start = CalendarUtils.toString(CalendarUtils.createCalendar(scheduledStart), SCHEDULED_FORMAT);
\r
275 result.setScheduleDate(scheduledStart);
\r
276 result.setItemHouseId(String.valueOf(rundownID));
\r
277 result.setItemTitle(String.format("%s %s %s", start, name, channel));
\r
279 // StringBuilder sb = new StringBuilder();
\r
280 // for (DBObject s : stories) {
\r
281 // BasicDBObject story = (BasicDBObject) s;
\r
282 // sb.append(story.getString(IOctopusAPI.PARENT_STORY_ID));
\r
283 // sb.append(" [" + story.getString(IOctopusAPI.FORMAT) + "] ");
\r
284 // sb.append(story.getString(IOctopusAPI.NAME));
\r
285 // sb.append("\r\n");
\r
288 StringBuilder sb = new StringBuilder();
\r
289 for (DBObject s : stories) {
\r
290 BasicDBObject story = (BasicDBObject) s;
\r
293 sb.append(story.getString(IOctopusAPI.PARENT_STORY_ID));
\r
294 sb.append(" [" + story.getString(IOctopusAPI.FORMAT) + "] ");
\r
295 sb.append(story.getString(IOctopusAPI.NAME));
\r
298 String content = story.getString(IOctopusAPI.SCRIPT_CONTENT);
\r
299 if (content != null) {
\r
300 content = content.replace("\r\n\r\n\r\n\r\n", "\r\n");
\r
301 content = content.replace("\r\n\r\n\r\n", "\r\n");
\r
302 content = content.replace("\r\n\r\n", "\r\n");
\r
303 sb.append(content);
\r
308 StoryArchive storyArchive = new StoryArchive();
\r
309 storyArchive.setMediaHouseId(result.getItemHouseId());
\r
310 storyArchive.setMediaTitle(clipName);
\r
311 storyArchive.setMediaDesc(sb.toString());
\r
312 result.addStoryArchive(storyArchive);
\r
313 storyArchive.addFileArchive(new FileArchive(clipName, duration));
\r
317 private List<BasicDBObject> queryClips() {
\r
318 DBCollection collection = db.getCollection(NEXIOCLIPS);
\r
319 BasicDBList agencies = new BasicDBList(filterAgencies);
\r
321 Calendar now = CalendarUtils.createZeroCalendar();
\r
322 QueryBuilder dateQueryBuilder = QueryBuilder.start(RECORDDATE).lessThan(now.getTime());
\r
323 QueryBuilder queryBuilder = QueryBuilder.start(EXTAGENCY).in(agencies).and(dateQueryBuilder.get());
\r
324 DBCursor cursor = collection.find(queryBuilder.get()).sort(RECORDDATE, -1);
\r
325 if (!cursor.hasNext())
\r
327 return ListUtils.cast(cursor.toArray());
\r
330 private void setAndCheck(int nexioPort, String nexioUserName, String nexioPassword, String archiveFtp, String archiveUserName, String archivePassword,
\r
331 String agencies, int limit, int nexioKillDateDays, String nexioAgency, IJobEngine jobEngine) throws Exception {
\r
332 db = NoSQLUtils.getNoSQLDB();
\r
334 logger.error(systemMarker, "Az NoSQL adatkezelő réteg nem elérhető.");
\r
335 throw new NullPointerException("Internal error, missing NoSQL DB reference.");
\r
338 if (jobEngine == null) {
\r
339 logger.error(systemMarker, "Az folyamatkezelő réteg nem elérhető.");
\r
340 throw new NullPointerException("Internal error, missing JobEngine reference.");
\r
342 manager = jobEngine.getItemManager();
\r
343 if (manager == null) {
\r
344 logger.error(systemMarker, "Az adatbáziskezelő réteg nem elérhető.");
\r
345 throw new NullPointerException("Internal error, missing ItemManager reference.");
\r
347 String nexioHost = System.getProperty("nexio.host");
\r
348 if (StringUtils.isBlank(nexioHost)) {
\r
349 logger.error(systemMarker, "A 'nexio.host' rendszer paraméter nem található.");
\r
350 throw new NullPointerException("System is not configured properly, 'jobengine.selenio.address' startup parameter missing.");
\r
352 check(nexioPort, "nexioPort");
\r
353 check(nexioUserName, "nexioUserName");
\r
354 check(nexioPassword, "nexioPassword");
\r
356 check(agencies, "filterAgencies");
\r
357 filterAgencies = agencies.split(",");
\r
359 check(limit, "limit");
\r
361 check(nexioKillDateDays, "nexioKillDateDays");
\r
362 this.nexioKillDateDays = nexioKillDateDays;
\r
363 check(nexioAgency, "nexioAgency");
\r
364 this.nexioAgency = nexioAgency;
\r
366 sourceUri = manager.createStoreUri(RemoteStoreProtocol.FTP, nexioHost);
\r
367 sourceUri.setPortNumber(nexioPort);
\r
368 sourceUri.setUserName(nexioUserName);
\r
369 sourceUri.setPassword(nexioPassword);
\r
370 if (sourceUri == null) {
\r
371 logger.error(systemMarker, "A forrás nem elérhető.");
\r
372 throw new NullPointerException("Internal error, missing 'sourceUri'.");
\r
375 check(archiveFtp, "archiveFtp");
\r
376 check(archiveUserName, "archiveUserName");
\r
377 check(archivePassword, "archivePassword");
\r
379 targetUri = manager.createStoreUri(new URI(archiveFtp));
\r
380 targetUri.setUserName(archiveUserName);
\r
381 targetUri.setPassword(archivePassword);
\r
382 if (targetUri == null) {
\r
383 logger.error(systemMarker, "A cél nem elérhető.");
\r
384 throw new NullPointerException("Internal error, missing 'targetUri'.");
\r
389 private void transferFile(String sourceFileName, String targetFileName) throws Exception {
\r
391 logger.info("Transfer clip {}", sourceFileName);
\r
393 sourceFtp = ((FtpDirectoryLister) sourceUri.getLister()).connect();
\r
394 targetFtp = ((FtpDirectoryLister) targetUri.getLister()).connect();
\r
396 if (!targetFtp.enterRemotePassiveMode())
\r
397 throw new Exception("!PASV");
\r
399 reply = sourceFtp.port(InetAddress.getByName(targetFtp.getPassiveHost()), targetFtp.getPassivePort());
\r
400 if (!FTPReply.isPositiveCompletion(reply))
\r
401 throw new Exception("PORT parancs válasza: " + sourceFtp.getReplyString());
\r
403 if (!sourceFtp.setFileType(FTP.BINARY_FILE_TYPE))
\r
404 throw new Exception("!SOURCE TYPE");
\r
406 reply = sourceFtp.retr(sourceFileName);
\r
407 if (!FTPReply.isPositivePreliminary(reply))
\r
408 throw new Exception("RETR parancs válasza: " + sourceFtp.getReplyString());
\r
410 if (!targetFtp.setFileType(FTP.BINARY_FILE_TYPE))
\r
411 throw new Exception("!TARGET TYPE");
\r
413 reply = targetFtp.stor(targetFileName);
\r
414 if (!FTPReply.isPositivePreliminary(reply))
\r
415 throw new Exception("STOR parancs válasza: " + sourceFtp.getReplyString());
\r
418 reply = sourceFtp.stat();
\r
419 if (!FTPReply.isPositiveCompletion(reply))
\r
420 throw new Exception("STAT parancs válasza: " + sourceFtp.getReplyString());
\r
422 //logger.info("Status: {}", sourceFtp.getReplyString());
\r
423 if (reply == 226) {
\r
426 Thread.sleep(1000);
\r
428 } catch (Exception e) {
\r
429 logger.catching(e);
\r
432 sourceUri.cleanUp();
\r
433 targetUri.cleanUp();
\r
438 private void transferMetadata(String fileName, BasicDBObject metadata) throws Exception {
\r
439 logger.info("Transfer metadata {}", fileName);
\r
440 OutputStream outStream = null;
\r
442 targetFtp = ((FtpDirectoryLister) targetUri.getLister()).connect();
\r
443 if (!targetFtp.changeWorkingDirectory(EscortFiles.STATUSFOLDER)) {
\r
444 targetFtp.makeDirectory(EscortFiles.STATUSFOLDER);
\r
445 if (!targetFtp.changeWorkingDirectory(EscortFiles.STATUSFOLDER))
\r
446 throw new Exception("!STATUSFOLDER");
\r
449 outStream = targetFtp.storeFileStream(fileName + JSON_EXT);
\r
450 if (outStream == null) {
\r
451 throw new NullPointerException("Can not open: " + fileName + JSON_EXT + " Reply:" + targetFtp.getReplyString());
\r
453 outStream.write(metadata.toString().getBytes(UTF_8));
\r
455 //targetFtp.changeToParentDirectory();
\r
456 } catch (Exception e) {
\r
457 logger.catching(e);
\r
460 if (outStream != null)
\r
462 targetUri.cleanUp();
\r
https://git.useribm.hu / mediacube.git / blob