1 package user.jobengine.server.steps;
\r
3 import java.io.OutputStream;
\r
4 import java.net.InetAddress;
\r
6 import java.util.ArrayList;
\r
7 import java.util.Calendar;
\r
8 import java.util.Date;
\r
9 import java.util.List;
\r
11 import org.apache.commons.lang.StringUtils;
\r
12 import org.apache.commons.net.ftp.FTP;
\r
13 import org.apache.commons.net.ftp.FTPClient;
\r
14 import org.apache.commons.net.ftp.FTPReply;
\r
15 import org.apache.logging.log4j.LogManager;
\r
16 import org.apache.logging.log4j.Logger;
\r
17 import org.apache.logging.log4j.Marker;
\r
19 import com.ibm.nosql.json.api.BasicDBObject;
\r
20 import com.ibm.nosql.json.api.DB;
\r
21 import com.ibm.nosql.json.api.DBCollection;
\r
22 import com.ibm.nosql.json.api.DBCursor;
\r
23 import com.ibm.nosql.json.api.DBObject;
\r
25 import user.commons.CalendarUtils;
\r
26 import user.commons.ListUtils;
\r
27 import user.commons.StoreUri;
\r
28 import user.commons.nosql.NoSQLUtils;
\r
29 import user.commons.octopus.IOctopusAPI;
\r
30 import user.commons.octopus.OctopusAPI;
\r
31 import user.commons.remotestore.FtpDirectoryLister;
\r
32 import user.commons.remotestore.RemoteStoreProtocol;
\r
33 import user.jobengine.db.IItemManager;
\r
34 import user.jobengine.server.IJobEngine;
\r
35 import user.jobengine.server.IJobRuntime;
\r
36 import user.jobengine.server.steps.MetadataTypeDetector.MetadataType;
\r
38 public class CopyForArchiveNEXIOMaterialsStep extends JobStep {
\r
39 private static final String SCHEDULED_FORMAT = "yyyy.MM.dd HH:mm";
\r
40 private static final Logger logger = LogManager.getLogger();
\r
41 private static final String UTF_8 = "utf-8";
\r
42 private static final String JSON_EXT = ".json";
\r
43 private static final String XML_EXT = ".xml";
\r
44 private static final String DURATION = "duration";
\r
45 private static final String MXFEXT = ".MXF";
\r
46 private static final String NEXIOCLIPS = "nexioclips";
\r
47 private static final String LONGNAMEID = "longnameid";
\r
48 private static final String ARCHIVEDRUNDOWNS = "archivedrundowns";
\r
49 private static final String ID = "id";
\r
50 private static final String MEDIATYPE = "Hír bejátszó";
\r
51 private OctopusAPI octopusAPI;
\r
52 private IItemManager manager;
\r
55 private FTPClient sourceFtp;
\r
56 private FTPClient targetFtp;
\r
57 private StoreUri sourceUri;
\r
58 private StoreUri targetUri;
\r
59 private int nexioKillDateDays;
\r
60 private String nexioAgency;
\r
61 private Marker systemMarker;
\r
62 private List<String> transferredFileNames = null;
\r
64 private int check(int value, String name) {
\r
66 logger.error(systemMarker, "A folyamat '{}' bemeneti paramétere 0.", name);
\r
67 throw new NullPointerException(String.format("System is not configured properly, missing '%s' input parameter.", name));
\r
72 private String check(String value, String name) {
\r
73 if (value == null) {
\r
74 logger.error(systemMarker, "A folyamat '{}' bemeneti paramétere üres.", name);
\r
75 throw new NullPointerException(String.format("System is not configured properly, missing '%s' input parameter.", name));
\r
80 private void copy(RundownArchive rundownArchive) throws Exception {
\r
81 for (StoryArchive storyArchive : rundownArchive.getStoryArchives()) {
\r
82 for (FileArchive fileArchive : storyArchive.getFileArchives()) {
\r
84 copyFile(fileArchive, rundownArchive, storyArchive);
\r
85 } catch (Exception e) {
\r
86 logger.error(systemMarker, "A '{}' clip archiválása sikertelen. A rendszer üzenete: {}", fileArchive.getFileName(), e.getMessage());
\r
92 private void copyFile(FileArchive fileArchive, RundownArchive rundownArchive, StoryArchive storyArchive) throws Exception {
\r
93 String origFileName = fileArchive.getFileName();
\r
94 String fileName = String.format("%s-%s", origFileName, rundownArchive.getItemHouseId());
\r
95 String videoFileName = fileName + MXFEXT;
\r
97 if (transferredFileNames == null)
\r
98 transferredFileNames = new ArrayList<>();
\r
100 //A mar letezo mozikat nem archivaljuk le ujra, csak a metaadatot
\r
101 long existingMediaId = 0;
\r
102 if (transferredFileNames.contains(origFileName)) {
\r
103 logger.info(systemMarker, "A '{}' file archiválásra felkészítése egy másik story kapcsán már megtörtént, ezért csak metaadat archiválás szükséges.",
\r
105 transferChunk(videoFileName);
\r
107 existingMediaId = ItemManagerExtensions.getExistingRundownMedia(manager, origFileName);
\r
108 if (existingMediaId == 0) {
\r
109 transferredFileNames.add(origFileName);
\r
110 transferFile(origFileName + MXFEXT, videoFileName);
\r
112 logger.info(systemMarker, "A '{}' file archiválása már megtörtént, ezért csak metaadat archiválás szükséges.", origFileName);
\r
113 transferChunk(videoFileName);
\r
117 BasicDBObject metadata = createMetadata(rundownArchive, storyArchive, fileArchive, existingMediaId);
\r
118 transferMetadata(videoFileName, metadata);
\r
119 createSourceKillDateFile(rundownArchive, origFileName);
\r
122 private BasicDBObject createMetadata(RundownArchive rundownArchive, StoryArchive storyArchive, FileArchive fileArchive, long existingMediaId) {
\r
123 BasicDBObject result = new BasicDBObject();
\r
124 result.put("itemHouseId", rundownArchive.getItemHouseId());
\r
125 result.put("itemTitle", rundownArchive.getItemTitle());
\r
126 result.put("itemDescription", rundownArchive.getItemDesc());
\r
127 result.put("userName", "mediacube");
\r
128 result.put("mediaHouseId", storyArchive.getMediaHouseId());
\r
129 result.put("mediaTitle", storyArchive.getMediaTitle());
\r
130 result.put("mediaDescription", storyArchive.getMediaDesc());
\r
131 result.put("mediaType", MEDIATYPE);
\r
132 result.put("duration", fileArchive.getDuration());
\r
133 result.put("existingMediaId", existingMediaId);
\r
137 private void createSourceKillDateFile(RundownArchive rundownArchive, String fileName) throws Exception {
\r
138 logger.info("Create killdate/agency for {}", fileName);
\r
139 OutputStream outStream = null;
\r
141 sourceFtp = ((FtpDirectoryLister) sourceUri.getLister()).connect();
\r
142 Calendar killDate = CalendarUtils.createCalendar(rundownArchive.getScheduleDate());
\r
143 killDate.add(Calendar.DAY_OF_YEAR, nexioKillDateDays);
\r
144 byte[] killDateFile = EscortFiles.createNEXIOKillDateFile(fileName, killDate.getTime(), null, nexioAgency);
\r
145 outStream = sourceFtp.storeFileStream(fileName + XML_EXT);
\r
146 if (outStream == null) {
\r
147 throw new NullPointerException("Can not open: " + fileName + XML_EXT + " Reply:" + sourceFtp.getReplyString());
\r
149 outStream.write(killDateFile);
\r
151 } catch (Exception e) {
\r
152 logger.catching(e);
\r
155 if (outStream != null)
\r
157 sourceUri.cleanUp();
\r
162 public Object[] execute(int nexioPort, String nexioUserName, String nexioPassword, String archiveFtp, String archiveUserName, String archivePassword,
\r
163 int daysBeforeNow, int nexioKillDateDays, String nexioAgency, IJobEngine jobEngine, IJobRuntime jobRuntime) throws Exception {
\r
164 systemMarker = jobRuntime.getMarker();
\r
165 setAndCheck(nexioPort, nexioUserName, nexioPassword, archiveFtp, archiveUserName, archivePassword, nexioKillDateDays, nexioAgency, jobEngine);
\r
166 octopusAPI = new OctopusAPI();
\r
167 Calendar scheduledDate = Calendar.getInstance();
\r
168 scheduledDate.add(Calendar.DAY_OF_YEAR, -1 * daysBeforeNow);
\r
169 List<DBObject> rundowns = octopusAPI.getRundowns(scheduledDate.getTime());
\r
170 if (rundowns == null) {
\r
171 logger.warn(systemMarker, "Nem található adástükör a {} napra.", CalendarUtils.toDateString(scheduledDate));
\r
175 processRundowns(rundowns);
\r
179 // private String getVersionedFileName(String fileName, String extension) throws Exception {
\r
180 // String result = fileName;
\r
182 // targetFtp = ((FtpDirectoryLister) targetUri.getLister()).connect();
\r
183 // FTPFile[] listFiles = targetFtp.listFiles();
\r
184 // List<String> fileNames = new ArrayList<>();
\r
185 // for (FTPFile ftpFile : listFiles) {
\r
186 // fileNames.add(ftpFile.getName());
\r
188 // while (fileNames.contains(result + extension)) {
\r
192 // } catch (Exception e) {
\r
193 // logger.catching(e);
\r
196 // targetUri.cleanUp();
\r
201 private FileArchive processMosObject(BasicDBObject rundown, BasicDBObject story, BasicDBObject mosObject) throws Exception {
\r
202 String mosID = mosObject.getString(IOctopusAPI.OBJ_ID);
\r
203 if (MetadataTypeDetector.GuessMetadataType(mosID) != MetadataType.OctopusPlaceholder) {
\r
204 logger.trace("Skipping MOS object {}", mosID);
\r
207 DBCollection clips = db.getCollection(NEXIOCLIPS);
\r
208 BasicDBObject clip = (BasicDBObject) clips.findOne(new BasicDBObject(LONGNAMEID, mosID));
\r
209 if (clip == null) {
\r
210 logger.debug("File {} NOT exists", mosID);
\r
212 //throw new Exception(String.format("File NOT exists %s", mosID));
\r
214 logger.debug("File {} exists", mosID);
\r
216 long duration = NoSQLUtils.asLong(clip, DURATION);
\r
218 if (duration == 0) {
\r
219 logger.debug("File {} exists with 0 frame length", mosID);
\r
222 if (duration == 1) {
\r
223 logger.debug("File {} exists with 1 frame length", mosID);
\r
226 return new FileArchive(mosID, duration);
\r
229 private RundownArchive processRundow(DBObject r) throws Exception {
\r
230 BasicDBObject rundown = (BasicDBObject) r;
\r
231 long rundownID = rundown.getLong(ID);
\r
232 logger.info("Processing rundown {} {}", rundownID, rundown.getString(IOctopusAPI.NAME));
\r
233 List<DBObject> stories = octopusAPI.getRundownFullStories(rundownID);
\r
234 if (stories == null)
\r
236 RundownArchive result = new RundownArchive();
\r
238 long id = NoSQLUtils.asLong(rundown, IOctopusAPI.ID);
\r
241 String name = NoSQLUtils.asString(NoSQLUtils.asDBObject(rundown, IOctopusAPI.RUNDOWN_TYPE), IOctopusAPI.NAME);
\r
242 if (StringUtils.isBlank(name))
\r
244 String channel = NoSQLUtils.asString(NoSQLUtils.asDBObject(rundown, IOctopusAPI.CHANNEL), IOctopusAPI.NAME);
\r
245 Date scheduledStart = rundown.getDate(IOctopusAPI.SCHEDULED_START);
\r
246 if (scheduledStart == null)
\r
248 String start = CalendarUtils.toString(CalendarUtils.createCalendar(scheduledStart), SCHEDULED_FORMAT);
\r
249 result.setScheduleDate(scheduledStart);
\r
250 result.setItemHouseId(String.valueOf(id));
\r
251 result.setItemTitle(String.format("%s %s %s", start, name, channel));
\r
253 for (DBObject s : stories) {
\r
254 StoryArchive storyArchive = processStory(rundown, s);
\r
255 if (storyArchive == null)
\r
257 result.addStoryArchive(storyArchive);
\r
262 private void processRundowns(List<DBObject> rundowns) {
\r
264 //TODO kiveni publikálás előtt
\r
265 //db.getCollection(ARCHIVEDRUNDOWNS).drop();
\r
266 List<BasicDBObject> archivedRundowns = queryArchivedRundowns();
\r
270 for (DBObject r : rundowns) {
\r
271 BasicDBObject rundown = (BasicDBObject) r;
\r
272 setProgress(index * 100 / rundowns.size());
\r
273 long rundownID = NoSQLUtils.asLong(rundown, IOctopusAPI.ID);
\r
274 String rundownName = NoSQLUtils.asString(rundown, IOctopusAPI.NAME);
\r
276 BasicDBObject currentRundownID = new BasicDBObject(IOctopusAPI.ID, rundownID);
\r
277 if (archivedRundowns != null && archivedRundowns.contains(currentRundownID)) {
\r
278 logger.info("Skipping archived rundown {} {}", rundownID, rundownName);
\r
282 RundownArchive rundownArchive = processRundow(r);
\r
283 if (rundownArchive == null || rundownArchive.isEmpty()) {
\r
284 logger.info("Skipping rundown {} {}", rundownID, rundownName);
\r
288 logger.info("Saving rundown {} {}", rundownID, rundownName);
\r
289 copy(rundownArchive);
\r
290 logger.info(systemMarker, "A '{}' tükör {}db bejátszójának archiválása sikeres volt", rundownArchive.getItemTitle(),
\r
291 rundownArchive.getStoryArchives().size());
\r
293 db.getCollection(ARCHIVEDRUNDOWNS).save(currentRundownID);
\r
294 } catch (Exception e) {
\r
295 logger.catching(e);
\r
296 logger.error(systemMarker,
\r
297 String.format("A %s %s tükör archiválása nem lehetséges, mert a annak ellenőrzése hibát jelzett. A rendszer üzenete: %s", rundownID,
\r
298 rundownName, e.getMessage()));
\r
305 private StoryArchive processStory(BasicDBObject rundown, DBObject s) throws Exception {
\r
306 BasicDBObject story = (BasicDBObject) s;
\r
307 String parentStoryID = story.getString(IOctopusAPI.PARENT_STORY_ID);
\r
308 if (StringUtils.isBlank(parentStoryID)) {
\r
309 logger.warn("Story parentStoryID is null: {}", story.toPrettyString(null));
\r
312 logger.debug("Processing story {}", parentStoryID);
\r
313 List<BasicDBObject> mosObjects = NoSQLUtils.asList(story, IOctopusAPI.MOS_OBJECTS);
\r
314 if (mosObjects == null)
\r
316 StoryArchive storyArchive = null;
\r
317 for (BasicDBObject mosObject : mosObjects) {
\r
318 FileArchive fileArchive = processMosObject(rundown, story, mosObject);
\r
319 if (fileArchive == null)
\r
321 if (storyArchive == null) {
\r
322 storyArchive = new StoryArchive();
\r
323 storyArchive.setMediaHouseId(parentStoryID);
\r
324 storyArchive.setMediaTitle(story.getString(IOctopusAPI.NAME));
\r
325 storyArchive.setMediaDesc(story.getString(IOctopusAPI.SCRIPT_CONTENT));
\r
328 storyArchive.addFileArchive(fileArchive);
\r
330 return storyArchive;
\r
334 private List<BasicDBObject> queryArchivedRundowns() {
\r
335 List<BasicDBObject> result = null;
\r
336 DBCollection collection = db.getCollection(ARCHIVEDRUNDOWNS);
\r
337 DBCursor find = collection.find(new BasicDBObject(), new BasicDBObject(IOctopusAPI._ID, 0).append(IOctopusAPI.ID, 1));
\r
338 if (find.hasNext())
\r
339 result = ListUtils.cast(find.toArray());
\r
343 private void setAndCheck(int nexioPort, String nexioUserName, String nexioPassword, String archiveFtp, String archiveUserName, String archivePassword,
\r
344 int nexioKillDateDays, String nexioAgency, IJobEngine jobEngine) throws Exception {
\r
345 db = NoSQLUtils.getNoSQLDB();
\r
347 logger.error(systemMarker, "Az NoSQL adatkezelő réteg nem elérhető.");
\r
348 throw new NullPointerException("Internal error, missing NoSQL DB reference.");
\r
351 if (jobEngine == null) {
\r
352 logger.error(systemMarker, "Az folyamatkezelő réteg nem elérhető.");
\r
353 throw new NullPointerException("Internal error, missing JobEngine reference.");
\r
355 manager = jobEngine.getItemManager();
\r
356 if (manager == null) {
\r
357 logger.error(systemMarker, "Az adatbáziskezelő réteg nem elérhető.");
\r
358 throw new NullPointerException("Internal error, missing ItemManager reference.");
\r
360 String nexioHost = System.getProperty("nexio.host");
\r
361 if (StringUtils.isBlank(nexioHost)) {
\r
362 logger.error(systemMarker, "A 'nexio.host' rendszer paraméter nem található.");
\r
363 throw new NullPointerException("System is not configured properly, 'jobengine.selenio.address' startup parameter missing.");
\r
365 check(nexioPort, "nexioPort");
\r
366 check(nexioUserName, "nexioUserName");
\r
367 check(nexioPassword, "nexioPassword");
\r
369 check(nexioKillDateDays, "nexioKillDateDays");
\r
370 this.nexioKillDateDays = nexioKillDateDays;
\r
371 check(nexioAgency, "nexioAgency");
\r
372 this.nexioAgency = nexioAgency;
\r
374 sourceUri = manager.createStoreUri(RemoteStoreProtocol.FTP, nexioHost);
\r
375 sourceUri.setPortNumber(nexioPort);
\r
376 sourceUri.setUserName(nexioUserName);
\r
377 sourceUri.setPassword(nexioPassword);
\r
378 if (sourceUri == null) {
\r
379 logger.error(systemMarker, "A forrás nem elérhető.");
\r
380 throw new NullPointerException("Internal error, missing 'sourceUri'.");
\r
383 check(archiveFtp, "archiveFtp");
\r
384 check(archiveUserName, "archiveUserName");
\r
385 check(archivePassword, "archivePassword");
\r
387 targetUri = manager.createStoreUri(new URI(archiveFtp));
\r
388 targetUri.setUserName(archiveUserName);
\r
389 targetUri.setPassword(archivePassword);
\r
390 if (targetUri == null) {
\r
391 logger.error(systemMarker, "A cél nem elérhető.");
\r
392 throw new NullPointerException("Internal error, missing 'targetUri'.");
\r
397 private void transferChunk(String fileName) throws Exception {
\r
398 logger.info("Transfer chunk {}", fileName);
\r
399 OutputStream outStream = null;
\r
401 targetFtp = ((FtpDirectoryLister) targetUri.getLister()).connect();
\r
402 outStream = targetFtp.storeFileStream(fileName);
\r
403 if (outStream == null) {
\r
404 throw new NullPointerException("Can not open: " + fileName + MXFEXT + " Reply:" + targetFtp.getReplyString());
\r
406 } catch (Exception e) {
\r
407 logger.catching(e);
\r
410 if (outStream != null)
\r
412 targetUri.cleanUp();
\r
416 private void transferFile(String sourceFileName, String targetFileName) throws Exception {
\r
418 logger.info("Transfer clip {}", sourceFileName);
\r
420 sourceFtp = ((FtpDirectoryLister) sourceUri.getLister()).connect();
\r
421 targetFtp = ((FtpDirectoryLister) targetUri.getLister()).connect();
\r
422 if (!targetFtp.enterRemotePassiveMode())
\r
423 throw new Exception("!PASV");
\r
425 reply = sourceFtp.port(InetAddress.getByName(targetFtp.getPassiveHost()), targetFtp.getPassivePort());
\r
426 if (!FTPReply.isPositiveCompletion(reply))
\r
427 throw new Exception("PORT parancs válasza: " + sourceFtp.getReplyString());
\r
429 if (!sourceFtp.setFileType(FTP.BINARY_FILE_TYPE))
\r
430 throw new Exception("!SOURCE TYPE");
\r
432 reply = sourceFtp.retr(sourceFileName);
\r
433 if (!FTPReply.isPositivePreliminary(reply))
\r
434 throw new Exception("RETR parancs válasza: " + sourceFtp.getReplyString());
\r
436 if (!targetFtp.setFileType(FTP.BINARY_FILE_TYPE))
\r
437 throw new Exception("!TARGET TYPE");
\r
439 reply = targetFtp.stor(targetFileName);
\r
440 if (!FTPReply.isPositivePreliminary(reply))
\r
441 throw new Exception("STOR parancs válasza: " + sourceFtp.getReplyString());
\r
444 reply = sourceFtp.stat();
\r
445 if (!FTPReply.isPositiveCompletion(reply))
\r
446 throw new Exception("STAT parancs válasza: " + sourceFtp.getReplyString());
\r
448 logger.info("Status: {}", sourceFtp.getReplyString());
\r
449 if (reply == 226) {
\r
452 Thread.sleep(1000);
\r
454 } catch (Exception e) {
\r
455 logger.catching(e);
\r
458 sourceUri.cleanUp();
\r
459 targetUri.cleanUp();
\r
464 private void transferMetadata(String fileName, BasicDBObject metadata) throws Exception {
\r
465 logger.info("Transfer metadata {}", fileName);
\r
466 OutputStream outStream = null;
\r
468 targetFtp = ((FtpDirectoryLister) targetUri.getLister()).connect();
\r
469 if (!targetFtp.changeWorkingDirectory(EscortFiles.STATUSFOLDER)) {
\r
470 targetFtp.makeDirectory(EscortFiles.STATUSFOLDER);
\r
471 if (!targetFtp.changeWorkingDirectory(EscortFiles.STATUSFOLDER))
\r
472 throw new Exception("!STATUSFOLDER");
\r
475 outStream = targetFtp.storeFileStream(fileName + JSON_EXT);
\r
476 if (outStream == null) {
\r
477 throw new NullPointerException("Can not open: " + fileName + JSON_EXT + " Reply:" + targetFtp.getReplyString());
\r
479 outStream.write(metadata.toString().getBytes(UTF_8));
\r
481 //targetFtp.changeToParentDirectory();
\r
482 } catch (Exception e) {
\r
483 logger.catching(e);
\r
486 if (outStream != null)
\r
488 targetUri.cleanUp();
\r
https://git.useribm.hu / mediacube.git / blob